Skip to content
Permalink
Browse files
Merge pull request #387 from EarthChen/feature/observer-compress
[Triple] Add observer compress test case
  • Loading branch information
guohao committed Oct 29, 2021
2 parents ab1c41e + 603ca67 commit b1c5dc98b6a71a6b7d7f3f4ab32fd92f32fce626
Showing 3 changed files with 156 additions and 1 deletion.
@@ -20,6 +20,15 @@ public interface PbGreeterManual {

StreamObserver<GreeterRequest> cancelBiStream2(StreamObserver<GreeterReply> replyStream);


StreamObserver<GreeterRequest> compressorBiStream(StreamObserver<GreeterReply> replyStream);


StreamObserver<GreeterRequest> clientCompressorBiStream(StreamObserver<GreeterReply> replyStream);


StreamObserver<GreeterRequest> serverCompressorBiStream(StreamObserver<GreeterReply> replyStream);

/**
* only use by query cancel result
*
@@ -2,6 +2,7 @@

import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.protocol.tri.ServerStreamObserver;
import org.apache.dubbo.sample.tri.GreeterReply;
import org.apache.dubbo.sample.tri.GreeterRequest;
import org.apache.dubbo.sample.tri.PbGreeter;
@@ -50,7 +51,6 @@ public void cancelServerStream(GreeterRequest request, StreamObserver<GreeterRep

@Override
public StreamObserver<GreeterRequest> cancelBiStream(StreamObserver<GreeterReply> replyStream) {
System.out.println("-----cancelBiStream thread=" + Thread.currentThread().getName());
RpcContext.getCancellationContext()
.addListener(context -> {
System.out.println("cancel--cancelBiStream");
@@ -105,6 +105,48 @@ public void onCompleted() {
};
}

@Override
public StreamObserver<GreeterRequest> compressorBiStream(StreamObserver<GreeterReply> replyStream) {
ServerStreamObserver<GreeterReply> replyServerStreamObserver = (ServerStreamObserver<GreeterReply>) replyStream;
replyServerStreamObserver.setCompression("gzip");
return getGreeterRequestStreamObserver(replyServerStreamObserver);
}

private StreamObserver<GreeterRequest> getGreeterRequestStreamObserver(StreamObserver<GreeterReply> streamObserver) {
return new StreamObserver<GreeterRequest>() {
@Override
public void onNext(GreeterRequest data) {
streamObserver.onNext(GreeterReply.newBuilder()
.setMessage(data.getName())
.build());
}

@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
streamObserver.onError(new IllegalStateException("Stream err"));
}

@Override
public void onCompleted() {
streamObserver.onCompleted();
}
};
}

@Override
public StreamObserver<GreeterRequest> clientCompressorBiStream(StreamObserver<GreeterReply> replyStream) {
ServerStreamObserver<GreeterReply> replyServerStreamObserver = (ServerStreamObserver<GreeterReply>) replyStream;
return getGreeterRequestStreamObserver(replyServerStreamObserver);
}

@Override
public StreamObserver<GreeterRequest> serverCompressorBiStream(StreamObserver<GreeterReply> replyStream) {
ServerStreamObserver<GreeterReply> replyServerStreamObserver = (ServerStreamObserver<GreeterReply>) replyStream;
replyServerStreamObserver.setCompression("gzip");
return getGreeterRequestStreamObserver(replyServerStreamObserver);
}

@Override
public GreeterReply queryCancelResult(GreeterRequest request) {
boolean canceled = cancelResultMap.getOrDefault(request.getName(), false);
@@ -5,6 +5,7 @@
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.protocol.tri.CancelableStreamObserver;
import org.apache.dubbo.rpc.protocol.tri.ClientStreamObserver;
import org.apache.dubbo.sample.tri.helper.StdoutStreamObserver;
import org.apache.dubbo.sample.tri.service.PbGreeterManual;
import org.junit.AfterClass;
@@ -111,6 +112,7 @@ public void onCompleted() {
(CancelableStreamObserver<GreeterRequest>) requestObserver;
for (int i = 0; i < n; i++) {
streamObserver.onNext(request);
streamObserver.cancel(new RuntimeException());
}
streamObserver.onCompleted();
Thread.sleep(2000);
@@ -163,6 +165,108 @@ public void onCompleted() {
Assert.assertEquals("true", reply.getMessage());
}

@Test
public void compressorBiStream() throws InterruptedException {
int n = 10;
CountDownLatch latch = new CountDownLatch(n);
final GreeterRequest request = GreeterRequest.newBuilder()
.setName("stream request")
.build();
StreamObserver<GreeterReply> observer = new CancelableStreamObserver<GreeterReply>() {
@Override
public void onNext(GreeterReply data) {
System.out.println(data);
latch.countDown();
}

@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
}

@Override
public void onCompleted() {
System.out.println("onCompleted");
}
};
final ClientStreamObserver<GreeterRequest> requestObserver =
(ClientStreamObserver<GreeterRequest>) delegateManual.compressorBiStream(observer);
requestObserver.setCompression("gzip");
for (int i = 0; i < n; i++) {
requestObserver.onNext(request);
}
requestObserver.onCompleted();
Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
}


@Test
public void clientCompressorBiStream() throws InterruptedException {
int n = 10;
CountDownLatch latch = new CountDownLatch(n);
final GreeterRequest request = GreeterRequest.newBuilder()
.setName("stream request")
.build();
StreamObserver<GreeterReply> observer = new CancelableStreamObserver<GreeterReply>() {
@Override
public void onNext(GreeterReply data) {
System.out.println(data);
latch.countDown();
}

@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
}

@Override
public void onCompleted() {
System.out.println("onCompleted");
}
};
final ClientStreamObserver<GreeterRequest> requestObserver =
(ClientStreamObserver<GreeterRequest>) delegateManual.clientCompressorBiStream(observer);
requestObserver.setCompression("gzip");
for (int i = 0; i < n; i++) {
requestObserver.onNext(request);
}
requestObserver.onCompleted();
Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
}

@Test
public void serverCompressorBiStream() throws InterruptedException {
int n = 10;
CountDownLatch latch = new CountDownLatch(n);
final GreeterRequest request = GreeterRequest.newBuilder()
.setName("stream request")
.build();
StreamObserver<GreeterReply> observer = new CancelableStreamObserver<GreeterReply>() {
@Override
public void onNext(GreeterReply data) {
System.out.println(data);
latch.countDown();
}

@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
}

@Override
public void onCompleted() {
System.out.println("onCompleted");
}
};
final ClientStreamObserver<GreeterRequest> requestObserver =
(ClientStreamObserver<GreeterRequest>) delegateManual.clientCompressorBiStream(observer);
for (int i = 0; i < n; i++) {
requestObserver.onNext(request);
}
requestObserver.onCompleted();
Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
}


@Test
public void stream() throws InterruptedException {

0 comments on commit b1c5dc9

Please sign in to comment.