SignalR Java Client 9.0.4: Memory leak in stream() method due to unbounded ReplaySubject

[ilyagorbunov]

Is there an existing issue for this?

• I have searched the existing issues

Describe the bug

Describe the bug

The SignalR Java client version 9.0.4 has a memory leak when using the stream() method for real-time data streaming. The issue is caused by the use of unbounded ReplaySubject.create() in two places:

1. InvocationRequest.java:14 - private final Subject<Object> pendingCall = ReplaySubject.create();
2. HubConnection.java:796 - ReplaySubject<T> subject = ReplaySubject.create(); (in the stream() method)

These unbounded replay subjects cache ALL emitted items indefinitely, causing memory to grow without bounds during long-running WebSocket connections with continuous data streams.

To Reproduce

Steps to reproduce the behavior:

1. Create a SignalR Java client connection
2. Use the stream() method to subscribe to a high-frequency data stream (e.g., real-time trading data)
3. Keep the connection running for extended periods (>1 hour)
4. Monitor memory usage - it will continuously grow until OutOfMemoryError occurs

Example code:

HubConnection hubConnection = HubConnectionBuilder.create("wss://example.com/hub")
    .withAccessTokenProvider(Single.just(accessToken))
    .build();

hubConnection.start().blockingAwait();

// This will cause memory leak over time
Observable<MarketData> stream = hubConnection.stream(MarketData.class, "StreamMarketData", "symbol");
stream.subscribe(data -> {
    // Process real-time data
});

Expected behavior

The streaming connection should maintain stable memory usage over time, not accumulating all historical data in memory.

Actual behavior

Memory usage grows continuously as ReplaySubject caches all emitted items. This leads to OutOfMemoryError after extended periods of streaming.

Workaround

Currently, we're forced to periodically disconnect and reconnect the SignalR connection every 5-15 minutes to clear the internal buffers and prevent OOM errors.

Suggested Fix

Replace unbounded ReplaySubject.create() with either:

1. Bounded replay: ReplaySubject.createWithSize(bufferSize)
2. PublishSubject for no caching: PublishSubject.create()
3. Make buffer size configurable

Environment

• SignalR Java Client version: 9.0.4
• Java version: 11/17
• Platform: Android
• RxJava version: 3.x

Stack trace

java.lang.OutOfMemoryError: Failed to allocate a 24 byte allocation with 1349584 free bytes and 1317KB until OOM
    at com.android.org.conscrypt.ConscryptEngine.newResult(ConscryptEngine.java:1363)
    at com.android.org.conscrypt.ConscryptEngine.unwrap(ConscryptEngine.java:933)
    ...
    at okhttp3.internal.ws.WebSocketReader.processNextFrame(WebSocketReader.kt:0)
    at okhttp3.internal.ws.RealWebSocket.loopReader(RealWebSocket.kt:293)

Additional context

This issue affects production applications using SignalR for real-time data streaming, particularly in financial trading applications where data streams are continuous and high-frequency.

Expected Behavior

No response

Steps To Reproduce

No response

Exceptions (if any)

No response

.NET Version

No response

Anything else?

No response

[BrennanConroy]

Nice find. Would you be interested in contributing a fix for this?

[ilyagorbunov]

Thank you for the quick response! Yes, I'd be happy to contribute a fix for this issue.

Proposed Solution

After analyzing the code, I believe the simplest and most effective fix is to replace the unbounded ReplaySubject.create() calls with bounded versions using ReplaySubject.createWithSize(1).

This will:

1. Limit the buffer to only the most recent value
2. Prevent unlimited memory growth during long-running streams
3. Maintain backward compatibility
4. Keep the same behavior for most use cases

Changes Required

The fix requires minimal changes in two files:

1. HubConnection.java:796

   // Before
   ReplaySubject<T> subject = ReplaySubject.create();
   
   // After  
   ReplaySubject<T> subject = ReplaySubject.createWithSize(1);

2. InvocationRequest.java:14

   // Before
   private final Subject<Object> pendingCall = ReplaySubject.create();
   
   // After
   private final Subject<Object> pendingCall = ReplaySubject.createWithSize(1);

Alternative Solutions Considered

1. PublishSubject - Would not cache any values, might break existing code expecting at least one cached value
2. BehaviorSubject - Similar to ReplaySubject(1) but requires an initial value
3. replay(1) operator - More complex implementation, same result

Testing

I can add tests to verify:

• Memory usage remains stable during long-running streams
• Existing functionality is preserved
• Late subscribers still receive the most recent value

Next Steps

Before I proceed with the PR, I have a few questions:

1. Do you agree with the ReplaySubject.createWithSize(1) approach, or would you prefer a different solution?
2. Are there any specific test scenarios you'd like me to include?
3. Should I add a configuration option to customize the buffer size, or is a fixed size of 1 acceptable?

I'm familiar with the contribution guidelines and will ensure the PR follows all requirements including signing the CLA.

Looking forward to your feedback!

[BrennanConroy]

Hmm, we probably should have used a UnicastSubject originally. That would be a behavior break now.

However, looking at this more closely, I don't think fixing the Stream method will fix the memory leak. We use a ReplaySubject in

aspnetcore/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/InvocationRequest.java

Line 14 in e336a0b

private final Subject<Object> pendingCall = ReplaySubject.create();

I think we should change this one to a UnicastSubject and then the internals of the Stream method should be refactored to not call subscribe on the Subject from getPendingCall until user code calls subscribe.