Questions tagged [system.reactive]

System.Reactive refers to the Reactive Extensions for .NET, also known as Rx. Rx provides developers with a reactive programming model over the generic IObservable<T> interface, as opposed to the traditional imperative programming model or the other reactive programming models that rely strictly on .NET Events or specific APIs.

Filter by
Sorted by
Tagged with
0 votes
1 answer
33 views

Processing batched events with RX Observables in Hosted Service

I have the following scenario: public class MyHostedService : BackgroundService { private readonly IServiceProvider _serviceProvider; private readonly IEventSource _eventSource; private ...
mnj's user avatar
  • 2,747
1 vote
2 answers
68 views

Rx - Scheduler that only allows n number of parallel operations at a time

I am trying to write a Scheduler that allows Limited Concurrency. Here is what I have written. public class ThreeAtATimeScheduler : IScheduler { private readonly SemaphoreSlim _semaphore = new ...
fahadash's user avatar
  • 3,221
0 votes
0 answers
34 views

How to create a Stream out of IObservable<byte>?

I've been thinking how to create a bridge between IObservable<byte> and Stream, but I'm completely lost. The use case for this is when a function needs you to provide a Stream, but you have your ...
SuperJMN's user avatar
  • 13.4k
0 votes
2 answers
70 views

Converting Stream to Observable

I've tried with this, but it seems to have concurrency issues. I don't fully understand what is wrong. public static IObservable<byte> ToObservable(this Stream stream, int bufferSize = 4096) { ...
SuperJMN's user avatar
  • 13.4k
0 votes
0 answers
68 views

How to cast from IObserver<IMessage[]> to IObserver<IMessage>

I'd like to know how can i safetly cast this varaible of type IObserver<IMessage[]> into IObserver .I'm using C# with .NetcoreApp versione public IDisposable Subscribe(IObserver<IMessage[]>...
EliaEmanuele's user avatar
0 votes
2 answers
97 views

How can I make async await the Reactive Extensions onNext, onError, onCompleted delegates

I have created the below extension where I was able to make an IObservable.Subscribe awaitable, but wrapping the whole subscription. A big issue in current async programming in C# and the code below ...
xplat's user avatar
  • 8,444
0 votes
0 answers
63 views

Using .net rx in regular REST API - does it make sense?

I recently heard about reactive programming and started to explore Observer C# pattern and rx library. Despite some tutorials, articles and example use cases I still can't think if there are any ...
T4under4's user avatar
  • 135
2 votes
1 answer
52 views

RX implementation of Save() action that would run immediately if 2s has passed since the previous execution or delay it until 2s has passed

var i = 0; var saveSource = Observable.Interval(TimeSpan.FromMilliseconds(200)).Select(x => i++); var throttledClicks = saveSource .Throttle(TimeSpan.FromMilliseconds(2000)) // Throttle for ...
user007's user avatar
  • 1,162
0 votes
0 answers
14 views

How to execute two methods async in RxJava?

How to execute two methods async? I have two methods call from menu click, and I need second method execute after first. But I can't understand why result.get() in makeMark() waits Thread.sleep(10000) ...
Slava's user avatar
  • 473
2 votes
2 answers
59 views

Is RX.net subject thread safe for subscribers?

I'm having a hard time finding information as google yields results about the safety of concurrently pushing data to the subject, this isn't my use case, i have a dedicated thread (running an infinite ...
Ronan Thibaudau's user avatar
0 votes
0 answers
31 views

Implementing request-response with Rx.NET

The code below simulates an HTTP-like protocol over a WebSocket connection. Each request expects a corresponding response. If a response is not received within a specified timeout interval, the ...
nop's user avatar
  • 5,043
0 votes
1 answer
56 views

Watching changes in children recursively with DynamicData

I'm creating a fully generic file explorer for AvaloniaUI. It has a view that shows a folder tree that you can navigate by expanding each directory. My goal is to make this view flexible enough to ...
SuperJMN's user avatar
  • 13.4k
0 votes
0 answers
65 views

Implementing non-overlapping calls to a method

I'm developing a websocket client wrapper. The third-party server I'm connecting to manages sessions, and it doesn't tolerate concurrent sessions. If the AuthenticateAsync method is called twice in a ...
nop's user avatar
  • 5,043
1 vote
1 answer
26 views

Creating an array of observables based off a condition and then reducing that array to one observable

I have two types of observables A & B, they return the same item type but are accessible by separate APIs. With a list of id's which can be for either A or B. These observables return data for a ...
stack's user avatar
  • 11
1 vote
2 answers
47 views

Executing a synchornous method, but with a timeout, using System.Reactive

What's the correct way to call the Read method with a timeout? public int Method() { return Read(); } I've tried with this, but my application is shutting down completely when a timeout ...
SuperJMN's user avatar
  • 13.4k
0 votes
0 answers
20 views

Execute OnNext on a Main Thread [duplicate]

I am trying to implement a kind of device driver in a reactive style. The idea is to have a thread fully occupied in exchanging and processing data from the device, and have a Subject to expose output ...
Improving's user avatar
0 votes
0 answers
22 views

How to cache results when using reactive subscription?

In a Blazor server-side app I have a component that is basically a dropdown list for postcodes. I use Reactive to make the calls to an endpoint that finds matching postcodes by search criteria. I ...
Ivan Debono's user avatar
1 vote
1 answer
68 views

Subscribe to observable get event *and* full history

I'm playing around with System.Reactive and looking to achieve the following: A timer fires an event every 3 seconds A filesystemwatcher fires events in case a file is changed (another app appending ...
Jochen van Wylick's user avatar
2 votes
1 answer
54 views

Rx.NET Buffer emit all items on Cancellation

I am using Rx.NET library and its method Buffer in the most simple manner like: observable.Buffer(TimeSpan.FromSeconds(5), 10); It works great except the case when cancellation token is activated. ...
Michal's user avatar
  • 1,354
0 votes
0 answers
19 views

How to make flattened stream of streams hot and connectible in RX.Net?

I have a reactive emitter that aggregates sources. When a source is added or removed it emits a new collection of observables in a stream. I use this to flatten it out. stream.Select(e => e.Merge())...
ZorgoZ's user avatar
  • 3,098
1 vote
1 answer
184 views

Subscribing to an Observable can be slow

I'm having performance issues with my my current project, and have narrowed it down to its use of observables. I have created the following console application to attempt to replicate this issue. var ...
JDChris100's user avatar
-1 votes
1 answer
39 views

Is there a visual in debug I can see the difference of using Dispose method on observable

Given: .Net 6 console application Visual Studio 2022 Is there a visual when debugging that I can see the difference of disposing the observable vs forgetting to dispose of it? https://dotnetfiddle.net/...
Rod's user avatar
  • 14.8k
2 votes
2 answers
64 views

How to implement Observable.Publish for an array of observables

I'm looking to implement the following function: public static IObservable<TResult> Publish<TSource, TResult>( this IObservable<TSource>[] sources, Func<IObservable<...
Bogey's user avatar
  • 5,072
0 votes
1 answer
31 views

Unexpected behavior in RxJava with observables constructed using AtomicIntegers

Here's the test case: import io.reactivex.rxjava3.core.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class MainTest { public static AtomicInteger ...
David Posner's user avatar
0 votes
0 answers
42 views

Running code within the same thread when cancellation is requested

I have a thread I am starting with Task.Run() - inside of it, I have an observable timer (Observable.Timer) that runs within the context of the same thread. As part of the thread execution, I am using ...
Den's user avatar
  • 16.7k
0 votes
1 answer
60 views

Pass async function to parameter with type of Action<X>? [duplicate]

The application has the following definition IDisposable IObservable<X>.Subscribe<X>(Action<X> onNext) It was passed an async function: async Task Process(X notification, ...
ca9163d9's user avatar
  • 27.8k
0 votes
0 answers
97 views

IAsyncEnumerable converted to IObservable (and Task) doesn't start

I have the following code in my BackgroundService: public async Task ExecuteAsync(CancellationToken cancellationToken) { await GetElements(cancellationToken) .ToObservable() ....
mnj's user avatar
  • 2,747
0 votes
1 answer
76 views

IObservable.Subscribe OnNext action is not blocking in blazor webassembly

I am using https://github.com/dotnet/reactive I use CombineLatest to combine observableA (emits items by timer) with observableB (insert manually). In OnNext method sometimes i insert new value to ...
nuclear sweet's user avatar
2 votes
2 answers
87 views

How to get MethodInfo of static method with generic parameters when method ambiguity occurs? [duplicate]

The method I would like to obtain is Observable.Return from System.Reactive. It's defined like this: public static IObservable<TResult> Return<TResult>(TResult value) { ... } I've ...
SuperJMN's user avatar
  • 13.4k
0 votes
1 answer
78 views

Disposable.Create async overload

I'm working on a message bus class library for Redis (StackExchange.Redis) and NATS (AlterNats). The SubscribeAsync method uses observers from System.Reactive for the callbacks. The problem is ...
nop's user avatar
  • 5,043
0 votes
0 answers
30 views

Making a method which uses Rx.NET non-throwable

I'm building a Redis PubSub Class Library and I would like to make SubscribeAsync non-throwable which means the Deserialize call should wrapped in a try/catch block. The thing is SubscribeAsync2 does ...
nop's user avatar
  • 5,043
0 votes
1 answer
560 views

Closure can be eliminated: method has overload to avoid closure creation

JetBrains Rider suggests the following over Disposable.Create: Closure can be eliminated: method has overload to avoid closure creation How do I fix it? public sealed class RedisSubscriber<TKey, ...
nop's user avatar
  • 5,043
0 votes
1 answer
68 views

view not updating using behaviorSubjects in Anguar project

in my Angular project I'm getting data from an api using a service and I'm saving data inside a behaviorSubject like so private categorySubject = new BehaviorSubject<number | null>(null); ...
Armando_pagano's user avatar
0 votes
0 answers
64 views

Filtering DynamicData source by ObservableCollection

I have a Vehicle entity as follows public class Vehicle { public string Name { get; } public int Id { get; } public Vehicle(int id, string name) { ...
James B's user avatar
  • 9,053
2 votes
2 answers
70 views

What is the correct way to take the most recent data in a reactivex ReplaySubject?

I'm new to reactivex (.NET) and I need help with the following problem. I have a ReplaySubject constructed by passing a time interval of 1 hour. It continuously receives data from a sensor. I want to ...
Alb's user avatar
  • 73
1 vote
2 answers
73 views

Skipping null values in a Scan method of an Observable collection

I have an IObservable of items with a timestamp. I use the Scan method to wrap each item and add a reference to the last valid wrapped item which was received. IObservable<IWrappedItem> ...
Shahar's user avatar
  • 655
1 vote
1 answer
41 views

TPL Dataflow with Rx: Observer misses messages

In the sample coding below, the source and IObservable and also part of a pipeline. The sink at the end of the pipeline gets all the messages. But the observer gets only the first and last messages. ...
DavidY's user avatar
  • 357
0 votes
1 answer
60 views

Rx Observable Sample operator that emits values at the beginning rather than the end of the period

RxJs includes the handy throttleTime operator that unlike .NET's Throttle operator can be configured to emit values from the source sequence at the beginning of the interval, rather than the end. Am I ...
Oliver Weichhold's user avatar
0 votes
1 answer
70 views

IObservable<Task<T>> to IObservable<T> that only produces the latest T

I have an observable stream IObservable<Task<T>> (call it stream A). From this stream, I wish to produce an IObservable<T> (stream B). If these were my only requirements, I believe ...
Alexander Høst's user avatar
2 votes
2 answers
49 views

(Rx.Net) Caching the data till an IObserver subscribes

I have a hot Observable (Rx.Net) that ticks when it detects a device connection to PC. I have a view that handles the display of these devices. Unfortunately, the View may not be displayed until after ...
resp78's user avatar
  • 1,464
0 votes
0 answers
57 views

Unit testing Rx Catch

I'm unit testing the following service and the issue is that I'm able to test only the happy path but not the unhappy path. So how do I test them? TestScheduler? http://introtorx.com/Content/v1.0....
nop's user avatar
  • 5,043
-1 votes
1 answer
51 views

Astonishing behavior with ReactiveX for .NET - Using Subject<int> to handle messages being pushed from a series of IObservable<int> objects [closed]

Originally I started out trying to solve the following question: I have a websocket client I wrote that needs to be periodically migrated using a well defined client protocol. The current ...
Daniel Green's user avatar
1 vote
1 answer
43 views

Print out Retry(5) attempts in Rx

I'm learning Rx and I wonder how I could print out the retries to the console, e.g. "Retry #1", "Retry #2", etc. I have seen some people using .Dump extension method that I cannot ...
nop's user avatar
  • 5,043
0 votes
1 answer
132 views

Make method more Rx like by replacing List<T> and TaskCompletionSource<T>

This is a request-response model (like HTTP) but over sockets/websockets. We know which response corresponds to which request by comparing the request IDs. The workflow is as following: Subscribe to ...
nop's user avatar
  • 5,043
1 vote
1 answer
95 views

Rx Observable on value types

I have an API receiving messages as structs in a callback method void OnReceive(ref T message) . That method receives those struct objects by reference to avoid copy. I am thinking to wrap that API ...
ddv's user avatar
  • 135
1 vote
1 answer
81 views

Reactive Merge Observables C#

I am trying to learn how to work with "System.Reactive" in my C# application. Now I am playing with the merge operations, but perhaps I misunderstood something. I have the following test ...
Franz Gsell's user avatar
  • 1,425
0 votes
0 answers
164 views

How to cause a delay with ReactiveUI.Validation in a ValidationRule

I am doing the following: this.ValidationRule(vm => vm.UserEntry, this.WhenAnyValue(vm => vm.UserEntry, IsEntryValidOrEmpty).Delay(TimeSpan.FromSeconds(5)), "Invalid entry.") Expecting ...
Xecrets's user avatar
  • 79
1 vote
2 answers
81 views

Emit the next value only if the previous value has completed something?

Say, I would like to have a stream that emits a new job 1 second after the previous job is done. The job should be emitted as soon as it's created, so the subscriber can track its progress. How could ...
hillin's user avatar
  • 1,632
0 votes
1 answer
145 views

Getting Issue with Source Cache in ReactiveUI

I have one issue in subscribing to source cache. Let me describe the problem. Lets say I have Test Class public class Test { public bool feature1 {get; set;} = false; public bool feature2 {get;...
Ritesh Kumar's user avatar
0 votes
1 answer
161 views

Visual Studio NuGet Package removes itself after reloading (Unity)

so I imported a library in my assembly project. Now I need to add some packages from the nuget launcher. As I install them my code gets fine, the namespace is found, and the package appears in my ...
Shainyyy's user avatar

1
2 3 4 5
69