Fix duplicate log messages for streaming upload (#260)

tracdap-libs/tracdap-lib-common/src/main/java/org/finos/tracdap/common/grpc/LoggingServerInterceptor.java

@@ -22,6 +22,7 @@
 import org.slf4j.LoggerFactory;
 
 public class LoggingServerInterceptor implements ServerInterceptor {
+
     private final Logger log;
 
     public LoggingServerInterceptor(Class<?> apiClass) {
@@ -32,64 +33,46 @@ public LoggingServerInterceptor(Class<?> apiClass) {
     public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
             ServerCall<ReqT, RespT> call,
             Metadata headers,
-            ServerCallHandler<ReqT, RespT> next
-    ) {
-        var loggingCall = new LoggingServerCall<>(call, log);
-        var listener = next.startCall(loggingCall, headers);
-        return new LoggingListener<>(listener, log, call.getMethodDescriptor());
-    }
+            ServerCallHandler<ReqT, RespT> next) {
 
+        var method = call.getMethodDescriptor();
+        var userInfo = AuthConstants.USER_INFO_KEY.get();
 
-    private static class LoggingServerCall<ReqT, RespT> extends ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT> {
-        private final Logger log;
+        log.info("API CALL START: [{}] [{} <{}>] ({})",
+                method.getBareMethodName(),
+                userInfo.getDisplayName(),
+                userInfo.getUserId(),
+                method.getType());
 
-        public LoggingServerCall(ServerCall<ReqT, RespT> delegate, Logger log) {
-            super(delegate);
+        var loggingCall = new LoggingServerCall<>(call);
+
+        return next.startCall(loggingCall, headers);
+    }
 
-            this.log = log;
+    private class LoggingServerCall<ReqT, RespT> extends ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT> {
+
+        public LoggingServerCall(ServerCall<ReqT, RespT> delegate) {
+            super(delegate);
         }
 
         @Override
         public void close(Status status, Metadata trailers) {
-            super.close(status, trailers);
 
             var method = getMethodDescriptor();
 
             if (status.isOk()) {
                 log.info("API CALL SUCCEEDED: [{}]", method.getBareMethodName());
-            } else {
+            }
+            else {
                 var grpcError = status.asRuntimeException();
                 // There is no GrpcErrorMapping.processError, because:
                 // 1) grpcError is always StatusRuntimeException
                 // 2) GrpcErrorMapping.processError passes through StatusRuntimeException
 
                 log.error("API CALL FAILED: [{}] {}", method.getBareMethodName(), grpcError.getMessage(), grpcError);
             }
-        }
-    }
-
-    private static class LoggingListener<ReqT> extends ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT> {
-        private final Logger log;
-        private final MethodDescriptor<ReqT, ?> method;
-
-        public LoggingListener(ServerCall.Listener<ReqT> delegate, Logger log, MethodDescriptor<ReqT, ?> method) {
-            super(delegate);
-
-            this.log = log;
-            this.method = method;
-        }
-
-        @Override
-        public void onMessage(ReqT message) {
-            var userInfo = AuthConstants.USER_INFO_KEY.get();
-
-            log.info("API CALL START: [{}] [{} <{}>] ({})",
-                    method.getBareMethodName(),
-                    userInfo.getDisplayName(),
-                    userInfo.getUserId(),
-                    method.getType());
 
-            super.onMessage(message);
+            delegate().close(status, trailers);
         }
     }
 }