cell_flow #

Codename: Mitosis · 1.0.0-rc.2

Rx-shaped operators for the Cell framework. Debounce, switchMap, zip, retry, buffer, and the complete reactive stream vocabulary — implemented as FlowInstructions on the same graph as package:cell.

🚀 Quick Start #

Installation #

Add both packages to your pubspec.yaml:

dependencies:
  cell: ^1.0.0-rc.4
  cell_flow: ^1.0.0-rc.4

Import once in your Dart file:

import 'package:cell_flow/cell_flow.dart'; // re-exports package:cell

Your First Pipeline (30 seconds) #

import 'package:cell_flow/cell_flow.dart';

final query = Cell.ingress<String>();

// Build a reactive search pipeline
final results = query.cell
    .map<String, String>(project: (q) => q.trim().toLowerCase())
    .filter<String>(test: (q) => q.length >= 2)
    .debounce<String>(duration: const Duration(milliseconds: 300))
    .asyncMapLatest<String, List<Hit>>(mapper: (q) => api.search(q));

// Observe results
Cell.observe(source: results.cell, effect: (p) {
  print('Found ${p.payload.length} hits');
});

// Emit user input
await query.emitAsync('Dart packages');

💡 New to cell_flow? Read the How to Start Guide for a complete 15-minute tutorial with 5 copy-paste examples.

🎯 What Problem Does Flow Solve? #

cell already gives you nodes (Cell), signals (Pulse), integrity (TestCell), and gates (Receptor). But everyday stream work — debouncing input, cancelling stale requests, batching events — would otherwise require custom Timers and one-off receptors.

cell_flow names those policies:

Key insight: The graph, locks, validation, and provenance stay in cell. Flow only decides which pulses leave and when.

🔧 Three Ways to Build Pipelines #

Way 1: Static Factories (Recommended for beginners) #

final box = Cell.ingress<String>();

final filtered = Flow.filter<String>(
  box.cell, 
  test: (q) => q.length >= 2,
);

final debounced = Flow.debounce<String>(
  filtered.cell,
  duration: const Duration(milliseconds: 300),
);

Best for: Multi-source operators (merge, zip, combineLatest), naming nodes for logs.

Way 2: Fluent Chaining (Most readable) #

final box = Cell.ingress<String>();

final results = box.cell
    .filter<String>(test: (q) => q.length >= 2)
    .debounce<String>(duration: const Duration(milliseconds: 300))
    .asyncMapLatest<String, List<String>>(mapper: api.search);

Best for: Linear pipelines, readability, quick prototyping.

⚠️ Note: Each fluent step creates a new Cell/Receptor. filter().map() = 2 cells.

Way 3: InstructionChain (One Receptor, many ops) #

final box = Cell.ingress<String>();

// Build a reusable policy - runs in ONE Receptor under ONE lock
final gate = MapValue<String, String>((s) => s.trim()) +
    Filter<String>((s) => s.isNotEmpty) +
    Distinct<String>();

final handle = gate.toHandle(source: box.cell);

Best for: Sync policies (trim, filter, tap), shared locks, unit testing gates.

💡 Key difference: + chains run in one Receptor under one lock. Fluent chains create multiple cells.

📊 Complete Operator Catalog (79+ Operators) #

Create Operators #

Transform Operators #

Filter / Take / Skip #

Flatten Operators (Inner Sequences) #

Combine Operators #

Time Operators #

Collect / Control Operators #

📖 Full documentation: FEATURES-Flow.md

🎓 Common Patterns #

Pattern 1: Search-as-You-Type #

final query = Cell.ingress<String>();

final results = query.cell
    .map<String, String>(project: (q) => q.trim().toLowerCase())
    .filter<String>(test: (q) => q.length >= 2)
    .distinct<String>()
    .debounce<String>(duration: const Duration(milliseconds: 300))
    .asyncMapLatest<String, List<Hit>>(mapper: api.search);

Why it works: asyncMapLatest cancels stale requests when the user types again.

Pattern 2: Form Validation #

final raw = Cell.ingress<String>();

final email = raw.cell
    .map<String, String>(project: (s) => s.trim().toLowerCase())
    .filter<String>(test: (s) => s.contains('@'))
    .distinct<String>()
    .asyncMap<String, User>(
      mapper: api.lookup,
      onError: (e, _) => log.warn(e),
    );

Pattern 3: Prevent Double-Checkout #

final taps = Cell.ingress<void>();

final receipts = taps.cell.exhaustMap<void, Receipt>(
  project: (_) => api.checkout(cart),
);

// While checkout runs, additional taps are ignored
await taps.emitAsync(null);  // Starts checkout
await taps.emitAsync(null);  // Ignored

Pattern 4: Batch Telemetry #

final events = Cell.ingress<Metric>();

final batches = events.cell
    .bufferWithTimeAndCount<Metric>(
      duration: const Duration(seconds: 5),
      count: 50,
    )
    .asyncMap<List<Metric>, void>(mapper: (batch) => api.flush(batch));

// Flushes when either 50 events arrive OR 5 seconds pass

Pattern 5: Combine Multiple Sources #

final price = Cell.ingress<int>();
final currency = Cell.ingress<String>();

final display = price.cell.withLatestFrom<int, String>(
  others: [currency.cell],
  combine: (p, latest) => '${latest.single} $p',
);

await price.emitAsync(20);       // Waits for currency
await currency.emitAsync('EUR'); // Now both have values
await price.emitAsync(25);       // → 'EUR 25'

⚠️ Gotcha: withLatestFrom waits until every other Cell has at least one value.

🧪 Testing Your Pipelines #

Test a Sync Chain (No Timer) #

import 'package:cell_flow/cell_flow.dart';
import 'package:test/test.dart';

void main() {
  test('sanitize drops blanks', () {
    final gate = MapValue<String, String>((s) => s.trim()) +
        Filter<String>((s) => s.isNotEmpty);
    
    final receptor = Receptor.instruction(gate);
    
    expect(receptor.call(Pulse('  x  '))?.payload, equals('x'));
    expect(receptor.call(Pulse('   ')), isNull);
  });
}

Test an Async Pipeline (With Timer) #

void main() {
  test('debounce emits last after silence', () async {
    final ingress = Cell.ingress<String>();
    
    final debounced = ingress.cell
        .debounce<String>(duration: const Duration(milliseconds: 100));
    
    final results = <String>[];
    Cell.observe(source: debounced.cell, effect: (p) {
      results.add(p.payload);
    });
    
    await ingress.emitAsync('a');
    await ingress.emitAsync('b');
    await ingress.emitAsync('c');
    
    // Wait past debounce duration
    await Future.delayed(const Duration(milliseconds: 150));
    
    expect(results, equals(['c']));
  });
}

📦 Package Layout #

package:cell_flow/cell_flow.dart
  part  src/flow.dart               Flow.* facade
  part  src/flow_core.dart          FlowInstruction, toHandle, operator +
  part  src/fluent_operator.dart    Cell / FlowHandle extensions
  import src/instruction/*.dart     operator classes

✅ Best practice: Application code should import only package:cell_flow/cell_flow.dart.

📚 Documentation #

🛠️ Runnable Examples #

Explore the example/ directory for complete, runnable demos:

Run any example:

cd packages/cell_flow
dart run example/filter_data_quality_demo.dart

⚙️ Important Rules #

1. A Cell does not complete. Use concat on inners (lists/futures/streams), not Cell-then-Cell.
2. reduce / scan are running folds, not terminal values.
3. Fluent filter().map() = 2 cells. Use Filter + MapValue for one Receptor.
4. Use + for sync policies. Keep timers and HTTP on the next node.
5. Keep cell and cell_flow on the same version line.

📌 Versioning #

Mitosis is the release codename (like an Android Studio animal). It is not part of the SemVer string. Stable on pub.flutter-io.cn will be 1.0.0.

⚠️ Status: Release Candidate (RC4). APIs may change. Not yet on pub.flutter-io.cn — use path or git dependencies. Verify behaviour against source.

📄 License #

Dual-licensed: MIT or Apache-2.0 — see LICENSE and file headers.

Copyright (c) 2025-Present Lee Man Hoi Simon. See AUTHORS.

ℹ️ Operator names follow ReactiveX vocabulary. Implementations are original Cell Flow; using switchMap as a name does not require an RxJS copyright header.

🔗 Links #

• 🏠 Repo: github.com/simon-m-lee/cell
• 📦 This package: packages/cell_flow
• 🐛 Issues: github.com/simon-m-lee/cell/issues
• 📖 Cell framework: packages/cell

Ready to build reactive pipelines? Start with the How to Start Guide →