{"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/294cab2f9b2e03af6b953820c7ab497c3c8b7ad9/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java#L775-L811","documentation":"A RuntimeException thrown by stream(Type returnType, Class<?> returnClass, String method, Object... args) when the connection is not CONNECTED. stream() registers a STREAM_INVOCATION and returns an Observable<T> of streamed items; starting a stream while disconnected would never yield items and leak the InvocationRequest.","triggerScenarios":"Calling connection.stream(ItemType.class, \"StreamMethod\", args) while getConnectionState() != CONNECTED — before start(), after stop(), or during reconnect.","commonSituations":"Subscribing to a stream on startup before the connection is established. Streaming after an onClosed event. Reconnect logic that resubscribes before state returns to CONNECTED.","solutions":["Ensure start() completed and state == CONNECTED before calling stream().","Guard the call with a state check.","Dispose stream subscriptions on disconnect and resubscribe only after reconnect."],"exampleFix":"// before\nObservable<Item> obs = connection.stream(Item.class, \"Watch\", id); // throws if not CONNECTED\nobs.subscribe(...);\n\n// after\nconnection.start().blockingAwait();\nif (connection.getConnectionState() == HubConnectionState.CONNECTED) {\n    Observable<Item> obs = connection.stream(Item.class, \"Watch\", id);\n    obs.subscribe(...);\n}","handlingStrategy":"validation","validationCode":"if (connection.getConnectionState() == HubConnectionState.CONNECTED) {\n    Observable<Item> obs = connection.stream(Item.class, \"method\", arg);\n    obs.subscribe(...);\n} else {\n    // defer subscription until connected\n}","typeGuard":"boolean canStream = connection.getConnectionState() == HubConnectionState.CONNECTED;","tryCatchPattern":"try {\n    Observable<Item> obs = connection.stream(Item.class, \"method\", arg);\n} catch (RuntimeException e) {\n    if (e.getMessage().contains(\"'stream' method cannot be called\")) {\n        // not connected; resubscribe on reconnect\n    } else { throw e; }\n}","preventionTips":["Gate stream() on CONNECTED state.","Dispose stream subscriptions on disconnect and resubscribe after reconnect.","Track active stream subscriptions in a manager that coordinates with connection lifecycle."],"tags":["signalr","java","lifecycle","stream","state-machine"],"analyzedSha":"294cab2f9b2e03af6b953820c7ab497c3c8b7ad9","analyzedAt":"2026-08-06T20:08:02.189Z","schemaVersion":2},"datasetVersion":"2026-08-06T23:17:07.152Z"}