Name: Rxdart Reactive Programming
Author: rakibul6670

Rxdart Reactive Programming | Skills Pool

Concern	How RxDart Handles It
State storage	`BehaviorSubject<T>` — caches last value, replays to new listeners
State updates	`.sink.add(data)` pushes new state
UI reactivity	`StreamBuilder(stream: rxObj.fileData)` rebuilds on new events
Error handling	`.sink.addError(err)` + centralized `ErrorMessageHandler`
Dependency injection	Global Rx objects in `api_acess.dart` — no `ChangeNotifier`, no `StateNotifier`
Stream composition	RxDart operators: `debounceTime`, `switchMap`, `combineLatest`, etc.

┌──────────────┐     ┌──────────────┐     ┌──────────────┐
│   API Layer  │────▶│   Rx Layer   │────▶│   UI Layer   │
│  (api.dart)  │     │  (rx.dart)   │     │ (screen.dart)│
│              │     │              │     │              │
│  Singleton   │     │  Extends     │     │ StreamBuilder │
│  Dio HTTP    │     │  RxResponseInt│    │  snapshot.data│
│  Endpoints.* │     │  BehaviorSubj│     │  snapshot.err │
└──────────────┘     └──────────────┘     └──────────────┘

// Without seed — starts empty
final subject = BehaviorSubject<int>();
subject.add(1);
subject.add(2);
subject.add(3);
subject.stream.listen(print); // prints 3 (latest cached value)

// With seed — starts with initial value
final seeded = BehaviorSubject<int>.seeded(0);
seeded.stream.listen(print); // prints 0 immediately

Property	Type	Description
`value`	`T`	Synchronously read the last emitted value
`valueOrNull`	`T?`	Same as `value` but nullable if no value emitted
`hasValue`	`bool`	`true` if at least one value has been emitted
`hasError`	`bool`	`true` if the last event was an error
`error`	`Object`	The last emitted error
`errorOrNull`	`Object?`	The last error or null
`stream`	`ValueStream<T>`	The observable stream
`sink`	`StreamSink<T>`	Add data/errors to the subject
`isClosed`	`bool`	Whether `close()` has been called
`hasListener`	`bool`	Whether any listener is subscribed

Method	Description
`add(T value)`	Push a new value to all listeners
`addError(Object error)`	Push an error to all listeners
`close()`	Close the subject permanently
`listen(onData, {onError, onDone})`	Subscribe to the stream

abstract class RxResponseInt<T> {
  T empty;                        // Reset value (usually {})
  BehaviorSubject<T> dataFetcher; // The core reactive stream

  RxResponseInt({required this.empty, required this.dataFetcher});

  dynamic handleSuccessWithReturn(T data) {
    dataFetcher.sink.add(data);   // Push data to UI
    return data;
  }

  dynamic handleErrorWithReturn(dynamic error) {
    log(error.toString());
    dataFetcher.sink.addError(error);
    ErrorMessageHandler.showErrorToast(error);
    return false;
  }

  void clean() => dataFetcher.sink.add(empty);  // Reset stream
  void dispose() => dataFetcher.close();         // Close permanently
}

@override
handleSuccessWithReturn(data) async {
  // Custom: save auth token, navigate, etc.
  String? token = data['data']['token'];
  DioSingleton.instance.update(token!);
  await appData.write(kKeyAccessToken, token);
  dataFetcher.sink.add(data);  // Always push to stream
  return true;
}

static String users() => "/users";

import 'dart:convert';
import 'package:dio/dio.dart';
import '../../../../networks/dio/dio.dart';
import '../../../../networks/endpoints.dart';
import '../../../../networks/exception_handler/data_source.dart';

final class GetUsersApi {
  static final GetUsersApi _singleton = GetUsersApi._internal();
  GetUsersApi._internal();
  static GetUsersApi get instance => _singleton;

  Future<Map> getUsersData() async {
    try {
      Response response = await getHttp(Endpoints.users());
      if (response.statusCode == 200 || response.statusCode == 201) {
        return json.decode(json.encode(response.data));
      } else {
        throw DataSource.DEFAULT.getFailure();
      }
    } catch (error) {
      rethrow;  // Let Rx layer handle errors
    }
  }
}

import 'package:rxdart/rxdart.dart';
import '../../../../helpers/error_message_handler.dart';
import '../../../../networks/rx_base.dart';
import 'api.dart';

final class GetUsersRx extends RxResponseInt {
  final api = GetUsersApi.instance;
  GetUsersRx({required super.empty, required super.dataFetcher});

  ValueStream get fileData => dataFetcher.stream;

  Future<bool> fetchUsers() async {
    try {
      Map data = await api.getUsersData();
      return await handleSuccessWithReturn(data);
    } catch (error) {
      return await handleErrorWithReturn(error);
    }
  }

  @override
  handleSuccessWithReturn(data) async {
    dataFetcher.sink.add(data);
    return true;
  }

  @override
  handleErrorWithReturn(error) {
    ErrorMessageHandler.showErrorToast(error);
    return false;
  }
}

GetUsersRx getUsersRxObj =
    GetUsersRx(empty: {}, dataFetcher: BehaviorSubject<Map>());

StreamBuilder(
  stream: getUsersRxObj.fileData,
  builder: (context, snapshot) {
    if (snapshot.connectionState == ConnectionState.waiting) {
      return const WaitingWidget();
    } else if (snapshot.hasData && snapshot.data != null) {
      return buildContent(snapshot.data!);
    } else {
      return const NotFoundWidget();
    }
  },
)

Future<bool> postLogin({required String email, required String password}) async {
  try {
    Map<String, dynamic> data = {"email": email, "password": password};
    Map resdata = await api.postLogIn(data);
    return await handleSuccessWithReturn(resdata);  // overridden below
  } catch (error) {
    return await handleErrorWithReturn(error);
  }
}

@override
handleSuccessWithReturn(data) async {
  String? token = data['data']['token'];
  int id = data['data']['user']['id'];
  DioSingleton.instance.update(token!);
  await appData.write(kKeyIsLoggedIn, true);
  await appData.write(kKeyUserID, id);
  await appData.write(kKeyAccessToken, token);
  dataFetcher.sink.add(data);
  performPostLoginActions();
  return true;
}

Operator	What It Does	Use Case
`debounceTime(Duration)`	Waits for a pause in events before emitting	Search-as-you-type input
`throttleTime(Duration)`	Limits emission rate	Scroll events, button spam
`switchMap(fn)`	Cancels previous inner stream on new event	API calls that replace previous
`distinctUnique()`	Filters out duplicate consecutive values	Avoid redundant rebuilds
`startWith(value)`	Prepends a value before stream events	Default/loading state
`scan(accumulator, seed)`	Reduces stream to accumulated value	Running totals, pagination
`combineLatest`	Combines latest values from multiple streams	Dashboard with multiple data sources
`mergeWith([streams])`	Merges multiple streams into one	Multiple event sources
`doOnData(fn)`	Side-effect on each data event	Logging, analytics
`doOnError(fn)`	Side-effect on each error	Error tracking
`onErrorReturn(value)`	Replace errors with a fallback value	Graceful degradation
`onErrorResumeNext(stream)`	Switch to fallback stream on error	Retry with different source
`whereNotNull()`	Filters out null values	Clean data pipeline
`delay(Duration)`	Delays each event	Artificial loading states

final searchSubject = BehaviorSubject<String>.seeded('');

// In initState or constructor:
searchSubject.stream
    .debounceTime(const Duration(milliseconds: 300))
    .where((query) => query.length >= 2)
    .distinct()  // skip if same query
    .switchMap((query) => Stream.fromFuture(api.search(query)))
    .listen((results) {
      searchResultsSubject.add(results);
    });

// In UI — push text changes to subject:
onChanged: (text) => searchSubject.add(text),

// Dashboard screen needs products AND profile data
Rx.combineLatest2(
  getProductsRxObj.fileData,
  profileRxObj.fileData,
  (products, profile) => {'products': products, 'profile': profile},
).listen((combined) {
  dashboardSubject.add(combined);
});

Class	Factory	Description
`CombineLatestStream`	`Rx.combineLatest2..9`	Combine latest values from 2-9 streams
`MergeStream`	`Rx.merge`	Merge multiple streams into one
`ConcatStream`	`Rx.concat`	Emit events from streams sequentially
`ForkJoinStream`	`Rx.forkJoin2..9`	Wait for all streams to complete, emit last values
`RetryStream`	`Rx.retry`	Retry a failing stream N times
`RetryWhenStream`	`Rx.retryWhen`	Retry with custom logic
`SwitchLatestStream`	`Rx.switchLatest`	Switch to latest inner stream
`RaceStream`	`Rx.race`	Emit only from the first stream to emit
`TimerStream`	`Rx.timer`	Emit a value after a delay
`DeferStream`	`Rx.defer`	Create stream lazily on subscription

Rule	Explanation
Use `Provider`/`Riverpod`/`Bloc`/`MobX`	This project uses RxDart exclusively
Call API from UI directly	Always go through the Rx controller
Use raw `Dio()`	Use `getHttp()`/`postHttp()` wrappers from `dio.dart`
Hardcode URLs in api.dart	Use `Endpoints.*` from `endpoints.dart`
Use `StreamController`	Use `BehaviorSubject` (replays last value)
Create Rx objects inside widgets	Register globally in `api_acess.dart`
Forget try/catch in Rx methods	Always wrap API calls in try/catch

Rule	Explanation
Extend `RxResponseInt`	All Rx controllers inherit from the base class
Override `handleSuccessWithReturn`	Customize success behavior per feature
Override `handleErrorWithReturn`	Customize error handling per feature
Use `ValueStream get fileData =>`	Expose the stream as a `ValueStream` getter
Register in `api_acess.dart`	One global instance per Rx class
Use `StreamBuilder` in UI	The pattern for observing Rx streams

Subject	Description	This Project Uses
`BehaviorSubject`	Caches latest value, replays to new listeners	✅ Primary
`ReplaySubject`	Caches ALL values, replays full history	❌ Not used (available if needed)

Feature	`StreamController`	`BehaviorSubject`
Replays last value to new listeners	❌ No	✅ Yes
Synchronous `.value` access	❌ No	✅ Yes
Broadcast by default	❌ No (single-sub)	✅ Yes
RxDart operators	❌ No	✅ Yes

Rxdart Reactive Programming

🔄 RxDart Reactive Programming Skill

🏗️ 1. Architecture: Why RxDart Instead of Provider/Riverpod

Rxdart Reactive Programming

🔄 RxDart Reactive Programming Skill

🏗️ 1. Architecture: Why RxDart Instead of Provider/Riverpod

📦 2. Core: `BehaviorSubject<T>` (Official API)

Key Properties

Key Methods

🧬 3. Base Class: `RxResponseInt<T>` (Project-Specific)

Override Pattern

🔌 4. Creating a GET Rx Controller (Complete Template)

Step 1: Endpoint (`lib/networks/endpoints.dart`)

Step 2: API Class (`data/rx_get_users/api.dart`)

Step 3: Rx Controller (`data/rx_get_users/rx.dart`)

Step 4: Register (`lib/networks/api_acess.dart`)

Step 5: UI (`presentation/users_screen.dart`)

📤 5. POST Rx Controller with Auth Token

🛠️ 6. RxDart Operators Reference (Use Instead of Provider Logic)

Commonly Useful Operators

Example: Search with Debounce

Example: Combining Streams

🔢 7. RxDart Stream Classes Reference

🧩 8. Subjects (Official API)

BehaviorSubject vs Dart StreamController

⚠️ 9. Rules & Common Mistakes

❌ NEVER Do

✅ ALWAYS Do

📚 10. Quick Reference: New API Checklist

Extract Errors

Frontend Patterns

Coding Standards

Coding Standards

Coding Standards

Frontend Patterns

Rxdart Reactive Programming

🔄 RxDart Reactive Programming Skill

🏗️ 1. Architecture: Why RxDart Instead of Provider/Riverpod

Rxdart Reactive Programming

🔄 RxDart Reactive Programming Skill

🏗️ 1. Architecture: Why RxDart Instead of Provider/Riverpod

📦 2. Core: BehaviorSubject<T> (Official API)

Key Properties

Key Methods

🧬 3. Base Class: RxResponseInt<T> (Project-Specific)

Override Pattern

🔌 4. Creating a GET Rx Controller (Complete Template)

Step 1: Endpoint (lib/networks/endpoints.dart)

Step 2: API Class (data/rx_get_users/api.dart)

Step 3: Rx Controller (data/rx_get_users/rx.dart)

Step 4: Register (lib/networks/api_acess.dart)

Step 5: UI (presentation/users_screen.dart)

📤 5. POST Rx Controller with Auth Token

🛠️ 6. RxDart Operators Reference (Use Instead of Provider Logic)

Commonly Useful Operators

Example: Search with Debounce

Example: Combining Streams

🔢 7. RxDart Stream Classes Reference

🧩 8. Subjects (Official API)

BehaviorSubject vs Dart StreamController

⚠️ 9. Rules & Common Mistakes

❌ NEVER Do

✅ ALWAYS Do

📚 10. Quick Reference: New API Checklist

Extract Errors

Frontend Patterns

Coding Standards

Coding Standards

Coding Standards

Frontend Patterns

📦 2. Core: `BehaviorSubject<T>` (Official API)

🧬 3. Base Class: `RxResponseInt<T>` (Project-Specific)

Step 1: Endpoint (`lib/networks/endpoints.dart`)

Step 2: API Class (`data/rx_get_users/api.dart`)

Step 3: Rx Controller (`data/rx_get_users/rx.dart`)

Step 4: Register (`lib/networks/api_acess.dart`)

Step 5: UI (`presentation/users_screen.dart`)