VDT.Core.Operators

Operators that process streams of published values from operand streams, allowing you to subscribe to the output stream for handling any sorts of events in a streamlined way. Create operand streams of the types you want to process, apply the required operators to those streams and subscribe to the results. Conceptually this is similar to piping observables, except an operand stream doesn't own the data it publishes - it's merely a conduit that's used for publishing, piping and subscribing. Below example uses a series of operators to ensure string values don't get published more than once every half a second and can be parsed as integers before subscribing to the resulting integer stream.

Features

• Asynchronous subscription to streams of values
• Asynchronous piping of values through a variety of operators to transform them
• Easily extensible with your own operators

Options

Each operand stream can be provided with an OperandStreamOptions object to specify how subscribers are interacted with.

• ReplayWhenSubscribing toggles the setting to publish all previously published values to a new subscriber when it is added
• ValueGenerator sets a method that will be executed when an operand stream is subscribed to, providing initial values
• ReplayValueGeneratorWhenSubscribing toggles the setting to determine when to execute the ValueGenerator
  • false to execute ValueGenerator only for the first subscriber
  • true to execute ValueGenerator for every subscriber

Please note that if ReplayWhenSubscribing and ReplayValueGeneratorWhenSubscribing are both false, any subscribers after the first will only receive values from ValueGenerator generated after they are subscribed.

Operators

• Debounce delays and throttles output values
• Filter discards output values based on a predicate
• Flatten subscribes to a stream of streams and outputs values published by the child streams
• GroupBy groups published values by using a key selector
• Iterate loops over received values of IEnumerable<T>
• Map transforms values
• Merge merges two or more streams
• QueueThrottle throttles output, queueing received values
• QueueZip outputs tuples of values published by two streams, queueing received values
• Throttle throttles output, discarding old received values
• Zip outputs tuples of values published by two streams, discarding old received values

Example

Below example uses a series of operators to ensure string values don't get published more than once every half a second and can be parsed as integers before subscribing to the resulting integer stream.

<input type="text" @oninput="async args => await valueStream.Publish(args.Value!.ToString()!)" /> @code { private readonly IOperandStream<string> valueStream = new OperandStream<string>(); protected override void OnAfterRender(bool firstRender) { if (firstRender) { valueStream .Debounce(500) .Map(value => new { IsValid = int.TryParse(value, out var result), Result = result }) .Filter(value => value.IsValid) .Map(value => value.Result) .Subscribe(value => { // Handle received integer values }); } } }

Custom operators

Although many common operators are available out of the box, it is simple to create your own by implementing either IOperator<TValue, TTransformedValue> to transform values without any initialization, or IOperator<TValue, TTransformedValue, TInitializationData> to add an initialization method with data to the operator. For ease of use, you can then also create extension methods for IOperandStream<TValue< that pipes values through your operator to the target stream.