Client Call Request Stream Issue

[ajdali]

Hi There,
I was encountering inconsistencies when calling a BiDirectional stream from the client side. I set up my request publisher as a Just to publish one value, and then I use combines AsyncStream helper method .values to wait for the response messages. However, most of the time, I never get a response back as it seems the connection closes prematurely. There are a few occurrences when the call is recieved and I am able to see the printed response. Am I setting up the request stream properly, or am I not supposed to use a Just publisher for the request stream? How do i ensure the connection is preserved without any of the streams terminating?

public struct LoginAuthStreamClient: Sendable {
    static func testAuth() async throws {
        let channel = try GRPCChannelManager.shared.getChannel() //setting up channel
        let loginAuthClient = PersistDotNetLib_AuthenticationNIOClient(channel: channel)
        //let loginAuthClient = PersistDotNetLib_AuthenticationNIOClient(channel: newChannel)//EchoServiceNIOClient(channel: channel)
        let requestStream: AnyPublisher<PersistDotNetLib_StreamAuthStatusRequest, Error> = fetchUploadAuthPublisher(authRequestType: .upload)
        let options = CallOptions(timeLimit: TimeLimit.timeout(.seconds(200)))
        
        let grpc = GRPCExecutor(callOptions: Just(options).eraseToAnyPublisher())
        
        for try await response in grpc.call(loginAuthClient.streamAuthStatus)(requestStream).values {
            print("Authentication Status: \(response.authStatus)")
            print("Authentication Reason: \(response.authMessage)")
        }
        //try! loginAuthClient.channel.close().wait()
    }
    
    static func fetchUploadAuthPublisher(authRequestType: AuthType) -> AnyPublisher<PersistDotNetLib_StreamAuthStatusRequest, Error> {
            let authRequest: PersistDotNetLib_StreamAuthStatusRequest
            if authRequestType == .upload {
                authRequest = .with {
                    $0.type = .upload
                }
            } else {
                authRequest = .with {
                    $0.type = .restore
                }
            }
            return Just(authRequest).setFailureType(to: Error.self).eraseToAnyPublisher()
        }
    }
}

[vyshane]

CombineGRPC will end the gRPC call when the client stream completes, and Just will send one value and complete immediately. If you want to keep the server stream open, you will need to ensure that your client steam does not complete.

If you want a single request -> response stream semantics, have you considered using a server stream RPC? For server streams, CombineGRPC will wait until the server ends the call.

[ajdali]

Thanks for the feedback. I ended up changing the requestPublisher to a passthroughsubject that never closes and sends messages whenever the provided value changes! :)

[ajdali]

@vyshane is there a process that you recommend for bidirectional streaming where the request stream can react to ui actions and then receive the responses? Do you have any examples with SwiftUI? In your examples you use publishers.sequence but I am trying to find how AnyPublisher can send messages based on UI actions.

Also, I am unable to use the CurrentValueSubject courtesy send method to send values to the request stream because we are needing to conform to AnyPublisher. Do you know of any workaround so that I can publish values to the request stream over time at my choosing?

[vyshane]

You should be able to use the eraseToAnyPublisher() method of any type that conforms to Publisher. I.e. use CurrentValueSubject, then erase its type to AnyPublisher before you pass it as a request to CombineGRPC.