dotnet/aspnetcore · error · RuntimeException

The 'stream' method cannot be called if the connection is…

Error message

The 'stream' method cannot be called if the connection is not active.

What it means

stream() opens a server-streaming invocation (StreamInvocationMessage). It checks the connection is CONNECTED before allocating the invocation id and ReplaySubject; otherwise it throws immediately under the state lock.

Solutions

Await start(): hub.start().blockingAwait(); then hub.stream(...).
Re-subscribe streams in the onReconnected callback, not onClosed, so the connection is already CONNECTED.
Gate stream calls behind your own connected flag flipped in onReconnected.

Example fix

// before
hub.start();
Observable<Integer> s = hub.stream(Integer.class, "Counter");

// after
hub.start().blockingAwait();
Observable<Integer> s = hub.stream(Integer.class, "Counter");

Defensive patterns

Strategy: validation

Validate before calling

if (hub.getConnectionState() != HubConnectionState.CONNECTED) {
  // do not open a stream; reconnect first
  return;
}
hub.stream(returnType, method, args);

Try / catch

try {
  hub.stream(returnType, method, args).subscribe(...);
} catch (RuntimeException e) {
  if (e.getMessage().contains("connection is not active")) {
    // retry stream setup after onReconnected
  } else throw e;
}

Prevention

Re-subscribe streams inside onReconnected, not onClosed.
Keep a registry of active streams so they can be re-opened automatically.
Await start() before the first stream().

When it happens

Trigger: Calling hub.stream(returnType, method, args) when the connection has not finished start(), has been stopped, or is reconnecting. The guard fires before the StreamInvocationMessage is sent.

Common situations: Subscribing to a stream from UI code that races start(); restarting streams after a reconnect without waiting for onReconnected; calling stream() inside onClosed before the new connection is up.

Related errors

AI-assisted analysis of dotnet/aspnetcore@3600ca084e (2026-08-11). Data as JSON: /api/errors/455c955b42894e66. Report an issue: GitHub.

Appendix: source

Thrown at src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java:793

     * @param returnType The expected return type of the stream items.
     * @param method The name of the server method to invoke.
     * @param args The arguments used to invoke the server method.
     * @param <T> The expected return type.
     * @return An observable that yields the streaming results from the server.
     */
    public <T> Observable<T> stream(Type returnType, String method, Object ... args) {
        Class<?> returnClass = Utils.typeToClass(returnType);
        return this.<T>stream(returnType, returnClass, method, args);
    }

    @SuppressWarnings("unchecked")
    private <T> Observable<T> stream(Type returnType, Class<?> returnClass, String method, Object ... args) {
        String invocationId;
        InvocationRequest irq;
        this.state.lock();
        try {
            if (this.state.getHubConnectionState() != HubConnectionState.CONNECTED) {
                throw new RuntimeException("The 'stream' method cannot be called if the connection is not active.");
            }

            ConnectionState connectionState = this.state.getConnectionStateUnsynchronized(false);
            invocationId = connectionState.getNextInvocationId();
            irq = new InvocationRequest(returnType, invocationId);
            connectionState.addInvocation(irq);

            AtomicInteger subscriptionCount = new AtomicInteger();
            ReplaySubject<T> subject = ReplaySubject.create();
            Subject<Object> pendingCall = irq.getPendingCall();
            pendingCall.subscribe(result -> {
                        subject.onNext(Utils.<T>cast(returnClass, result));
                    }, error -> subject.onError(error),
                    () -> subject.onComplete());

            Observable<T> observable = subject.doOnSubscribe((subscriber) -> subscriptionCount.incrementAndGet());
            sendInvocationMessage(method, args, invocationId, true);
            return observable.doOnDispose(() -> {

View on GitHub (pinned to 3600ca084e)