{"record":{"id":"455c955b42894e66","repo":"dotnet/aspnetcore","slug":"the-stream-method-cannot-be-called-if-the-connec","errorCode":null,"errorMessage":"The 'stream' method cannot be called if the connection is not active.","messagePattern":"The 'stream' method cannot be called if the connection is not active\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java","lineNumber":793,"sourceCode":"     * @param returnType The expected return type of the stream items.\n     * @param method The name of the server method to invoke.\n     * @param args The arguments used to invoke the server method.\n     * @param <T> The expected return type.\n     * @return An observable that yields the streaming results from the server.\n     */\n    public <T> Observable<T> stream(Type returnType, String method, Object ... args) {\n        Class<?> returnClass = Utils.typeToClass(returnType);\n        return this.<T>stream(returnType, returnClass, method, args);\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    private <T> Observable<T> stream(Type returnType, Class<?> returnClass, String method, Object ... args) {\n        String invocationId;\n        InvocationRequest irq;\n        this.state.lock();\n        try {\n            if (this.state.getHubConnectionState() != HubConnectionState.CONNECTED) {\n                throw new RuntimeException(\"The 'stream' method cannot be called if the connection is not active.\");\n            }\n\n            ConnectionState connectionState = this.state.getConnectionStateUnsynchronized(false);\n            invocationId = connectionState.getNextInvocationId();\n            irq = new InvocationRequest(returnType, invocationId);\n            connectionState.addInvocation(irq);\n\n            AtomicInteger subscriptionCount = new AtomicInteger();\n            ReplaySubject<T> subject = ReplaySubject.create();\n            Subject<Object> pendingCall = irq.getPendingCall();\n            pendingCall.subscribe(result -> {\n                        subject.onNext(Utils.<T>cast(returnClass, result));\n                    }, error -> subject.onError(error),\n                    () -> subject.onComplete());\n\n            Observable<T> observable = subject.doOnSubscribe((subscriber) -> subscriptionCount.incrementAndGet());\n            sendInvocationMessage(method, args, invocationId, true);\n            return observable.doOnDispose(() -> {","sourceCodeStart":775,"sourceCodeEnd":811,"githubUrl":"https://github.com/dotnet/aspnetcore/blob/3600ca084e9c8b5f4174fc5e747f4c52d2100806/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java#L775-L811","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nhub.start();\nObservable<Integer> s = hub.stream(Integer.class, \"Counter\");\n\n// after\nhub.start().blockingAwait();\nObservable<Integer> s = hub.stream(Integer.class, \"Counter\");","handlingStrategy":"validation","validationCode":"if (hub.getConnectionState() != HubConnectionState.CONNECTED) {\n  // do not open a stream; reconnect first\n  return;\n}\nhub.stream(returnType, method, args);","typeGuard":null,"tryCatchPattern":"try {\n  hub.stream(returnType, method, args).subscribe(...);\n} catch (RuntimeException e) {\n  if (e.getMessage().contains(\"connection is not active\")) {\n    // retry stream setup after onReconnected\n  } else throw e;\n}","preventionTips":["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()."],"tags":["signalr","java","state","streaming","race-condition"],"backgroundTag":null,"analyzedSha":"3600ca084e9c8b5f4174fc5e747f4c52d2100806","analyzedAt":"2026-08-11T16:32:30.678Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}