Revert "synchronized blocks were replaced with ReentrantLock in ReadStreamCall"

This reverts commit 0ed2460.

Author: Igor Melnichenko

core/src/main/java/tech/ydb/core/impl/call/ReadStreamCall.java

@@ -3,7 +3,6 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.ReentrantLock;
 
 import javax.annotation.Nullable;
 
@@ -30,7 +29,6 @@ public class ReadStreamCall<ReqT, RespT> extends ClientCall.Listener<RespT> impl
 
     private final String traceId;
     private final ClientCall<ReqT, RespT> call;
-    private final ReentrantLock callLock = new ReentrantLock();
     private final GrpcStatusHandler statusConsumer;
     private final ReqT request;
     private final Metadata headers;
@@ -58,40 +56,34 @@ public CompletableFuture<Status> start(Observer<RespT> observer) {
             throw new IllegalStateException("Read stream call is already started");
         }
 
-        callLock.lock();
-
-        try {
-            call.start(this, headers);
-            call.request(1);
-            if (logger.isTraceEnabled()) {
-                logger.trace("ReadStreamCall[{}] --> {}", traceId, TextFormat.shortDebugString((Message) request));
-            }
-            call.sendMessage(request);
-            // close stream by client side
-            call.halfClose();
-        } catch (Throwable t) {
+        synchronized (call) {
             try {
-                call.cancel(null, t);
-            } catch (Throwable ex) {
-                logger.error("ReadStreamCall[{}] got exception while canceling", traceId, ex);
+                call.start(this, headers);
+                call.request(1);
+                if (logger.isTraceEnabled()) {
+                    logger.trace("ReadStreamCall[{}] --> {}", traceId, TextFormat.shortDebugString((Message) request));
+                }
+                call.sendMessage(request);
+                // close stream by client side
+                call.halfClose();
+            } catch (Throwable t) {
+                try {
+                    call.cancel(null, t);
+                } catch (Throwable ex) {
+                    logger.error("ReadStreamCall[{}] got exception while canceling", traceId, ex);
+                }
+
+                statusFuture.completeExceptionally(t);
             }
-
-            statusFuture.completeExceptionally(t);
-        } finally {
-            callLock.unlock();
         }
 
         return statusFuture;
     }
 
     @Override
     public void cancel() {
-        callLock.lock();
-
-        try {
+        synchronized (call) {
             call.cancel("Cancelled on user request", new CancellationException());
-        } finally {
-            callLock.unlock();
         }
     }
 
@@ -103,23 +95,18 @@ public void onMessage(RespT message) {
             }
             observerReference.get().onNext(message);
             // request delivery of the next inbound message.
-            callLock.lock();
-
-            try {
+            synchronized (call) {
                 call.request(1);
-            } finally {
-                callLock.unlock();
             }
         } catch (Exception ex) {
             statusFuture.completeExceptionally(ex);
-            callLock.lock();
 
             try {
-                call.cancel("Canceled by exception from observer", ex);
+                synchronized (call) {
+                    call.cancel("Canceled by exception from observer", ex);
+                }
             } catch (Throwable th) {
                 logger.error("ReadStreamCall[{}] got exception while canceling", traceId, th);
-            } finally {
-                callLock.unlock();
             }
         }
     }