Reactive programming extensions and advanced RxJS utilities for the ziegel framework. Provides enhanced operators, reactive patterns, BLoC architecture, and subscription management utilities.

Reactive Extensions: Advanced reactive programming with custom operators, BLoC patterns, and enhanced RxJS utilities for complex data streams.

🚀 Overview

@breadstone/ziegel-rx provides:

• Enhanced RxJS Operators: Custom operators for common reactive patterns
• BLoC Architecture: Business Logic Component pattern implementation
• Subscription Management: Advanced subscription handling and cleanup
• Reactive Extensions: Extensions for observables and subjects
• Stream Utilities: Countdown, observe, and transformation utilities
• Pattern Implementations: Reactive design patterns for enterprise applications

📦 Installation

npm install @breadstone/ziegel-rx
# or
yarn add @breadstone/ziegel-rx

🧩 Features & Usage Examples

Enhanced RxJS Operators

import { notNull, empty } from '@breadstone/ziegel-rx';
import { map, filter } from 'rxjs/operators';

// Filter out null values from streams
const dataStream$ = source$.pipe(
  notNull(), // Removes null and undefined values
  map((data) => data.value),
);

// Empty operator for placeholder streams
const emptyStream$ = empty<string>();

BLoC Architecture Pattern

import { Bloc, BlocObserver, Transition } from '@breadstone/ziegel-rx';

// Define events and states
interface CounterEvent {
  type: 'increment' | 'decrement';
}

interface CounterState {
  count: number;
}

// Implement BLoC
class CounterBloc extends Bloc<CounterEvent, CounterState> {
  constructor() {
    super({ count: 0 });
  }

  mapEventToState(event: CounterEvent): CounterState {
    const currentState = this.state;
    switch (event.type) {
      case 'increment':
        return { count: currentState.count + 1 };
      case 'decrement':
        return { count: currentState.count - 1 };
      default:
        return currentState;
    }
  }
}

// Use BLoC
const counterBloc = new CounterBloc();
counterBloc.add({ type: 'increment' });

BLoC Observer

import { BlocObserver, Transition } from '@breadstone/ziegel-rx';

class AppBlocObserver extends BlocObserver {
  onTransition(transition: Transition<any, any>): void {
    console.log('State transition:', transition);
  }

  onError(error: Error): void {
    console.error('BLoC error:', error);
  }
}

// Set global observer
Bloc.observer = new AppBlocObserver();

Subscription Extensions

import { CompositeUnsubscribable, subscribeOnce, once } from '@breadstone/ziegel-rx';

// Composite subscription management
const subscriptions = new CompositeUnsubscribable();

subscriptions.add(stream1$.subscribe((value) => console.log('Stream 1:', value)));
subscriptions.add(stream2$.subscribe((value) => console.log('Stream 2:', value)));

// Unsubscribe from all at once
subscriptions.unsubscribe();

// Subscribe once to an observable
subscribeOnce(dataStream$, (value) => {
  console.log('Received once:', value);
});

// Alternative syntax
once(dataStream$, (value) => {
  console.log('Received once:', value);
});

Observable Extensions

import { observe, toBehaviorSubject, toReplaySubject, toSubject } from '@breadstone/ziegel-rx';
import { from } from 'rxjs';

// Convert arrays to observables with observe
const arrayObservable$ = observe([1, 2, 3, 4, 5]);

// Convert observables to different subject types
const source$ = from([1, 2, 3]);

const behaviorSubject = toBehaviorSubject(source$, 0); // With initial value
const replaySubject = toReplaySubject(source$, 3); // Replay last 3 values
const subject = toSubject(source$);

Countdown Utilities

import { countdown, countdown2 } from '@breadstone/ziegel-rx';

// Countdown from 10 to 0
countdown(10).subscribe((count) => {
  console.log('Countdown:', count);
});

// Advanced countdown with custom interval
countdown2(30, 1000).subscribe((count) => {
  console.log('30 second countdown:', count);
});

import { IReactiveCommand } from '@breadstone/ziegel-rx';

// Implement reactive commands for user interactions
class SaveCommand implements IReactiveCommand {
  execute() {
    // Command execution logic
  }

  canExecute() {
    // Command availability logic
  }
}

Enhanced Operators

import { customOperators } from '@breadstone/ziegel-rx';
import { of } from 'rxjs';

// Use enhanced operators for complex data transformations
const stream$ = of(1, 2, 3)
  .pipe
  // Custom operators for enterprise scenarios
  ();

Observable Patterns

import { reactivePatterns } from '@breadstone/ziegel-rx';

// Implement reactive data patterns
const dataStream$ = reactivePatterns.createDataStream();

📚 Package import points

📚 Package import points

import {
  // Subscription Management
  CompositeUnsubscribable,

  // Extensions
  countdown,
  countdown2,
  observe,
  subscribeOnce,
  once,
  toBehaviorSubject,
  toReplaySubject,
  toSubject,

  // Operators
  empty,
  notNull,

  // BLoC Pattern
  Bloc,
  BlocObserver,
  Transition,
  EventStreamClosedException,
} from '@breadstone/ziegel-rx';

📚 API Documentation

Related Packages

• @breadstone/ziegel-core: Foundation utilities and reactive patterns
• rxjs: Core reactive programming library

License

MIT

Issues

Please report bugs and feature requests in the Issue Tracker

---

Part of the ziegel Enterprise TypeScript Framework