lindonglin commented on issue #14629:
URL: https://github.com/apache/dubbo/issues/14629#issuecomment-2331520063
@cnzakii Hello, the following code can reproduce the phenomenon I
described。 After the provider is restarted, the consumer runs normally without
error, but the provider does not actually receive the message.
```proto
syntax = "proto3";
option java_multiple_files = true;
option java_package = "com.test.someutil.tri";
option java_outer_classname = "HelloWorldProto";
option objc_class_prefix = "HLW";
package helloworld;
// The request message containing the user's name.
message HelloRequest {
string name = 1;
}
// The response message containing the greetings
message HelloReply {
string message = 1;
}
```
```java
public interface IStreamGreeter {
StreamObserver<HelloRequest> sayHello(StreamObserver<HelloReply>
replyObserver);
}
```
```java
public class IStreamGreeterImpl implements IStreamGreeter {
@Override
public StreamObserver<HelloRequest> sayHello(StreamObserver<HelloReply>
replyObserver) {
return new StreamObserver<>() {
@Override
public void onNext(HelloRequest helloRequest) {
System.out.println("onNext receive request name:" +
helloRequest.getName());
replyObserver.onNext(HelloReply.newBuilder().setMessage("ok").build());
}
@Override
public void onError(Throwable cause) {
System.out.println("onError");
replyObserver.onError(cause);
}
@Override
public void onCompleted() {
System.out.println("onComplete");
replyObserver.onCompleted();
}
};
}
}
```
**provider main**
```java
public class DubboTripleStreamProvider {
public static void main(String[] args) throws InterruptedException {
ServiceConfig<IStreamGreeter> service = new ServiceConfig<>();
service.setInterface(IStreamGreeter.class);
service.setRef(new IStreamGreeterImpl());
service.setProtocol(new ProtocolConfig(CommonConstants.TRIPLE,
9002));
service.setApplication(new ApplicationConfig("stream-provider"));
service.setRegistry(new
RegistryConfig("zookeeper://127.0.0.1:2181"));
service.export();
System.out.println("dubbo service started");
new CountDownLatch(1).await();
}
}
```
**consumer main**
```java
public class DubboTripleStreamConsumer {
public static void main(String[] args) throws InterruptedException,
IOException {
ReferenceConfig<IStreamGreeter> ref = new ReferenceConfig<>();
ref.setInterface(IStreamGreeter.class);
ref.setCheck(false);
ref.setProtocol(CommonConstants.TRIPLE);
ref.setLazy(true);
ref.setTimeout(100000);
ref.setApplication(new ApplicationConfig("stream-consumer"));
ref.setUrl("tri://127.0.0.1:9002");
final IStreamGreeter iStreamGreeter = ref.get();
System.out.println("dubbo ref started");
try {
StreamObserver<HelloRequest> streamObserver =
iStreamGreeter.sayHello(new StreamObserver<>() {
@Override
public void onNext(HelloReply reply) {
System.out.println(reply.getMessage());
}
@Override
public void onError(Throwable throwable) {
System.out.println("onError:" + throwable.getMessage());
}
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
});
int i = 0;
while (true) {
streamObserver.onNext(HelloRequest.newBuilder()
.setName("hello " + i++)
.build());
Thread.sleep(5000L);
}
} catch (Throwable t) {
t.printStackTrace();
}
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
