client never gets notified when server goes away

[apbthere]

The client never gets notified when server goes away. It just waits indefinitely. The following unit test creates a simple server that streams 200 messages to the client. Client will call Object.wait() on server socket effectively hanging the server after receiving 2 messages and then wait for 20 more messages or it will time out in 10 seconds. The client doesn't throw any exceptions nor signals onError.

package com.example.demo;

import java.util.Date;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import com.poc.protobuf.UnitRequest;
import com.poc.protobuf.UnitResponse;
import com.poc.protobuf.UnitService;
import com.poc.protobuf.UnitServiceClient;
import com.poc.protobuf.UnitServiceServer;

import io.netty.buffer.ByteBuf;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.rpc.rsocket.RequestHandlingRSocket;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.server.NettyContextCloseable;
import io.rsocket.transport.netty.server.TcpServerTransport;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class DefaultSimpleServiceTests {

class DefaultSimpleService implements UnitService {

@Override
public Flux<UnitResponse> requestStream(UnitRequest unitRequest, ByteBuf metadata) {
	String command = unitRequest.getRequestCommandMessage();
	
	return Flux.range(1, 200)
			.map(i -> UnitResponse.newBuilder()
					.setMessageNumber(i)
					.setResponseMessage(i + " Srever is processsing " + command + " command")
					.build());
}
}

  @Test
  public void test1() throws Exception {
	  UnitServiceServer serviceServer = new UnitServiceServer(new DefaultSimpleService(), Optional.empty(), Optional.empty());
	  
	    NettyContextCloseable serverSocket = RSocketFactory.receive()
	        .acceptor(
	            (setup, sendingSocket) ->
	                Mono.just(new RequestHandlingRSocket(serviceServer)))
	        .transport(TcpServerTransport.create(8801))
	        .start()
	        .block();
	    
	    RSocket rSocket = RSocketFactory.connect().transport(TcpClientTransport.create(8801)).start().block();

	    UnitServiceClient client = new UnitServiceClient(rSocket);
	    
	    	CountDownLatch latch = new CountDownLatch(22);
	    	
	    	client.requestStream(UnitRequest.newBuilder().setRequestCommandMessage("Give me some data!").build())
	    			.subscribe(new Subscriber<UnitResponse>() {

				private Subscription subscription;

				@Override
				public void onSubscribe(Subscription s) {
					this.subscription = s;
					s.request(1);
				}

				@Override
				public void onNext(UnitResponse t) {
					System.out.println("Received message " + t.getResponseMessage());
					latch.countDown();
					
					if (latch.getCount() < 20) {
						System.out.println("Killing server now...");
						try {
							// this will halt the thread causing server to disappear 
							serverSocket.wait();
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
					
					subscription.request(1);
				}

				@Override
				public void onError(Throwable t) {
					System.out.println("Error detected! " + t);
				}

				@Override
				public void onComplete() {
					System.out.println("Stream completed!");
				}
			} );
	    	
	    	latch.await(10, TimeUnit.SECONDS);
	    	System.out.println("Finished at " + new Date().toString());
  }
}

proto file

syntax = "proto3";

package com.harris.atom.poc;

import "google/protobuf/empty.proto";

option java_package = "com.poc.protobuf";
option java_outer_classname = "UnitServiceProto";
option java_multiple_files = true;

service UnitService {

// Single Request / Streaming Response
rpc RequestStream (UnitRequest) returns (stream UnitResponse) {}
}

message UnitRequest {
string requestCommandMessage = 1;
}

message UnitResponse {
string responseMessage = 1;
int32 messageNumber = 2;
}

[robertroeser]

@mostroverkhov does your fix to rsocket help with this?

[mostroverkhov]

@robertroeser It helps, but https://github.com/reactor/reactor-netty/issues/495 has to be resolved also