Bump from reactor 2.3.5.RELEASE to 2.7.8 (#830)

* Bump from reactor 2.3.5.RELEASE to 2.7.8

Signed-off-by: Sergio <[email protected]>

* Simplification

Signed-off-by: Sergio <[email protected]>

---------

Signed-off-by: Sergio <[email protected]>

Author: champel

examples/src/main/java/io/dapr/examples/OpenTelemetryConfig.java

@@ -89,7 +89,7 @@ public static OpenTelemetry createOpenTelemetry() {
    * Converts current OpenTelemetry's context into Reactor's context.
    * @return Reactor's context.
    */
-  public static reactor.util.context.Context getReactorContext() {
+  public static reactor.util.context.ContextView getReactorContext() {
     return getReactorContext(Context.current());
   }
 

examples/src/main/java/io/dapr/examples/pubsub/BulkPublisher.java

@@ -71,7 +71,7 @@ public static void main(String[] args) throws Exception {
           System.out.println("Going to publish message : " + message);
         }
         BulkPublishResponse<?> res = client.publishEvents(PUBSUB_NAME, TOPIC_NAME, "text/plain", messages)
-            .subscriberContext(getReactorContext()).block();
+            .contextWrite(getReactorContext()).block();
         System.out.println("Published the set of messages in a single call to Dapr");
         if (res != null) {
           if (res.getFailedEntries().size() > 0) {

examples/src/main/java/io/dapr/examples/pubsub/PublisherWithTracing.java

@@ -63,7 +63,7 @@ public static void main(String[] args) throws Exception {
           client.publishEvent(
               PUBSUB_NAME,
               TOPIC_NAME,
-              message).subscriberContext(getReactorContext()).block();
+              message).contextWrite(getReactorContext()).block();
           System.out.println("Published message: " + message);
 
           try {

examples/src/main/java/io/dapr/examples/tracing/InvokeClient.java

@@ -67,7 +67,7 @@ public static void main(String[] args) throws Exception {
                 InvokeMethodRequest sleepRequest = new InvokeMethodRequest(SERVICE_APP_ID, "proxy_sleep")
                     .setHttpExtension(HttpExtension.POST);
                 return client.invokeMethod(sleepRequest, TypeRef.get(Void.class));
-              }).subscriberContext(getReactorContext()).block();
+              }).contextWrite(getReactorContext()).block();
         }
       }
     }

examples/src/main/java/io/dapr/examples/tracing/TracingDemoMiddleServiceController.java

@@ -58,7 +58,7 @@ public Mono<byte[]> echo(
     InvokeMethodRequest request = new InvokeMethodRequest(INVOKE_APP_ID, "echo")
         .setBody(body)
         .setHttpExtension(HttpExtension.POST);
-    return client.invokeMethod(request, TypeRef.get(byte[].class)).subscriberContext(getReactorContext(context));
+    return client.invokeMethod(request, TypeRef.get(byte[].class)).contextWrite(getReactorContext(context));
   }
 
   /**
@@ -71,7 +71,7 @@ public Mono<byte[]> echo(
   public Mono<Void> sleep(@RequestAttribute(name = "opentelemetry-context") Context context) {
     InvokeMethodRequest request = new InvokeMethodRequest(INVOKE_APP_ID, "sleep")
         .setHttpExtension(HttpExtension.POST);
-    return client.invokeMethod(request, TypeRef.get(byte[].class)).subscriberContext(getReactorContext(context)).then();
+    return client.invokeMethod(request, TypeRef.get(byte[].class)).contextWrite(getReactorContext(context)).then();
   }
 
 }

sdk-actors/src/main/java/io/dapr/actors/client/DaprGrpcClient.java

@@ -29,7 +29,7 @@
 import io.grpc.stub.StreamObserver;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
-import reactor.util.context.Context;
+import reactor.util.context.ContextView;
 
 import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
@@ -65,7 +65,7 @@ public Mono<byte[]> invoke(String actorType, String actorId, String methodName,
             .setMethod(methodName)
             .setData(jsonPayload == null ? ByteString.EMPTY : ByteString.copyFrom(jsonPayload))
             .build();
-    return Mono.subscriberContext().flatMap(
+    return Mono.deferContextual(
         context -> this.<DaprProtos.InvokeActorResponse>createMono(
             it -> intercept(context, client).invokeActor(req, it)
         )
@@ -109,7 +109,7 @@ public void start(final Listener<RespT> responseListener, final Metadata metadat
    * @param client GRPC client for Dapr.
    * @return Client after adding interceptors.
    */
-  private static DaprGrpc.DaprStub intercept(Context context, DaprGrpc.DaprStub client) {
+  private static DaprGrpc.DaprStub intercept(ContextView context, DaprGrpc.DaprStub client) {
     return GrpcWrapper.intercept(context, client);
   }
 

sdk-tests/pom.xml

@@ -131,7 +131,7 @@
       <dependency>
           <groupId>io.projectreactor</groupId>
           <artifactId>reactor-core</artifactId>
-          <version>3.3.11.RELEASE</version>
+          <version>3.5.0</version>
           <scope>test</scope>
       </dependency>
     <dependency>

sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeIT.java

@@ -101,7 +101,7 @@ public void testInvokeTimeout() throws Exception {
                     .block(Duration.ofMillis(10))).getMessage();
             long delay = System.currentTimeMillis() - started;
             assertTrue(delay <= 500);  // 500 ms is a reasonable delay if the request timed out.
-            assertEquals("Timeout on blocking read for 10 MILLISECONDS", message);
+            assertEquals("Timeout on blocking read for 10000000 NANOSECONDS", message);
         }
     }
 

sdk-tests/src/test/java/io/dapr/it/methodinvoke/http/MethodInvokeIT.java

@@ -118,7 +118,7 @@ public void testInvokeTimeout() throws Exception {
             }).getMessage();
             long delay = System.currentTimeMillis() - started;
             assertTrue(delay <= 200);  // 200 ms is a reasonable delay if the request timed out.
-            assertEquals("Timeout on blocking read for 10 MILLISECONDS", message);
+            assertEquals("Timeout on blocking read for 10000000 NANOSECONDS", message);
         }
     }
 

sdk-tests/src/test/java/io/dapr/it/tracing/grpc/TracingIT.java

@@ -79,7 +79,7 @@ public void testInvoke() throws Exception {
             try (Scope scope = span.makeCurrent()) {
                 SleepRequest req = SleepRequest.newBuilder().setSeconds(1).build();
                 client.invokeMethod(daprRun.getAppName(), "sleepOverGRPC", req.toByteArray(), HttpExtension.POST)
-                    .subscriberContext(getReactorContext())
+                    .contextWrite(getReactorContext())
                     .block();
             }
         }