Proposed gRPC, Java server API for bidirectional stream widgets

.gitignore

@@ -46,7 +46,8 @@ workspaces/
 devopsLocal/
 pyclient/venv
 pyclient/dist
-pyclient/*.egg-info
+.venv/
+*.egg-info/
 
 jenkins/failures
 

Integrations/src/main/java/io/deephaven/integrations/python/PythonDeephavenSession.java

@@ -43,7 +43,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 /**
@@ -87,7 +86,7 @@ public PythonDeephavenSession(
         scope = pythonEvaluator.getScope();
         executionContext.getQueryLibrary().importClass(org.jpy.PyObject.class);
         try (final SafeCloseable ignored = executionContext.open()) {
-            module = (PythonScriptSessionModule) PyModule.importModule("deephaven.server.script_session")
+            module = (PythonScriptSessionModule) PyModule.importModule("deephaven_internal.script_session")
                     .createProxy(CallableKind.FUNCTION, PythonScriptSessionModule.class);
         }
         scriptFinder = new ScriptFinder(DEFAULT_SCRIPT_PATH);
@@ -116,7 +115,7 @@ public PythonDeephavenSession(
         evaluator = null;
         this.scope = (PythonScope<PyObject>) scope;
         try (final SafeCloseable ignored = executionContext.open()) {
-            module = (PythonScriptSessionModule) PyModule.importModule("deephaven.server.script_session")
+            module = (PythonScriptSessionModule) PyModule.importModule("deephaven_internal.script_session")
                     .createProxy(CallableKind.FUNCTION, PythonScriptSessionModule.class);
         }
         scriptFinder = null;

authorization/src/main/java/io/deephaven/auth/codegen/impl/ObjectServiceAuthWiring.java

@@ -11,6 +11,7 @@
 import io.deephaven.auth.ServiceAuthWiring;
 import io.deephaven.proto.backplane.grpc.FetchObjectRequest;
 import io.deephaven.proto.backplane.grpc.ObjectServiceGrpc;
+import io.deephaven.proto.backplane.grpc.StreamRequest;
 import io.grpc.ServerServiceDefinition;
 
 /**
@@ -30,6 +31,12 @@ default ServerServiceDefinition intercept(ObjectServiceGrpc.ObjectServiceImplBas
 
         serviceBuilder.addMethod(ServiceAuthWiring.intercept(
                 service, "FetchObject", null, this::onMessageReceivedFetchObject));
+        serviceBuilder.addMethod(ServiceAuthWiring.intercept(
+                service, "MessageStream", this::onCallStartedMessageStream, this::onMessageReceivedMessageStream));
+        serviceBuilder.addMethod(ServiceAuthWiring.intercept(
+                service, "OpenMessageStream", null, this::onMessageReceivedOpenMessageStream));
+        serviceBuilder.addMethod(ServiceAuthWiring.intercept(
+                service, "NextMessageStream", null, this::onMessageReceivedNextMessageStream));
 
         return serviceBuilder.build();
     }
@@ -43,14 +50,77 @@ default ServerServiceDefinition intercept(ObjectServiceGrpc.ObjectServiceImplBas
      */
     void onMessageReceivedFetchObject(AuthContext authContext, FetchObjectRequest request);
 
+    /**
+     * Authorize a request to open a client-streaming rpc MessageStream.
+     *
+     * @param authContext the authentication context of the request
+     * @throws io.grpc.StatusRuntimeException if the user is not authorized to invoke MessageStream
+     */
+    void onCallStartedMessageStream(AuthContext authContext);
+
+    /**
+     * Authorize a request to MessageStream.
+     *
+     * @param authContext the authentication context of the request
+     * @param request the request to authorize
+     * @throws io.grpc.StatusRuntimeException if the user is not authorized to invoke MessageStream
+     */
+    void onMessageReceivedMessageStream(AuthContext authContext, StreamRequest request);
+
+    /**
+     * Authorize a request to OpenMessageStream.
+     *
+     * @param authContext the authentication context of the request
+     * @param request the request to authorize
+     * @throws io.grpc.StatusRuntimeException if the user is not authorized to invoke OpenMessageStream
+     */
+    void onMessageReceivedOpenMessageStream(AuthContext authContext, StreamRequest request);
+
+    /**
+     * Authorize a request to NextMessageStream.
+     *
+     * @param authContext the authentication context of the request
+     * @param request the request to authorize
+     * @throws io.grpc.StatusRuntimeException if the user is not authorized to invoke NextMessageStream
+     */
+    void onMessageReceivedNextMessageStream(AuthContext authContext, StreamRequest request);
+
     class AllowAll implements ObjectServiceAuthWiring {
         public void onMessageReceivedFetchObject(AuthContext authContext, FetchObjectRequest request) {}
+
+        public void onCallStartedMessageStream(AuthContext authContext) {}
+
+        public void onMessageReceivedMessageStream(AuthContext authContext, StreamRequest request) {}
+
+        public void onMessageReceivedOpenMessageStream(AuthContext authContext,
+                StreamRequest request) {}
+
+        public void onMessageReceivedNextMessageStream(AuthContext authContext,
+                StreamRequest request) {}
     }
 
     class DenyAll implements ObjectServiceAuthWiring {
         public void onMessageReceivedFetchObject(AuthContext authContext, FetchObjectRequest request) {
             ServiceAuthWiring.operationNotAllowed();
         }
+
+        public void onCallStartedMessageStream(AuthContext authContext) {
+            ServiceAuthWiring.operationNotAllowed();
+        }
+
+        public void onMessageReceivedMessageStream(AuthContext authContext, StreamRequest request) {
+            ServiceAuthWiring.operationNotAllowed();
+        }
+
+        public void onMessageReceivedOpenMessageStream(AuthContext authContext,
+                StreamRequest request) {
+            ServiceAuthWiring.operationNotAllowed();
+        }
+
+        public void onMessageReceivedNextMessageStream(AuthContext authContext,
+                StreamRequest request) {
+            ServiceAuthWiring.operationNotAllowed();
+        }
     }
 
     class TestUseOnly implements ObjectServiceAuthWiring {
@@ -61,5 +131,31 @@ public void onMessageReceivedFetchObject(AuthContext authContext, FetchObjectReq
                 delegate.onMessageReceivedFetchObject(authContext, request);
             }
         }
+
+        public void onCallStartedMessageStream(AuthContext authContext) {
+            if (delegate != null) {
+                delegate.onCallStartedMessageStream(authContext);
+            }
+        }
+
+        public void onMessageReceivedMessageStream(AuthContext authContext, StreamRequest request) {
+            if (delegate != null) {
+                delegate.onMessageReceivedMessageStream(authContext, request);
+            }
+        }
+
+        public void onMessageReceivedOpenMessageStream(AuthContext authContext,
+                StreamRequest request) {
+            if (delegate != null) {
+                delegate.onMessageReceivedOpenMessageStream(authContext, request);
+            }
+        }
+
+        public void onMessageReceivedNextMessageStream(AuthContext authContext,
+                StreamRequest request) {
+            if (delegate != null) {
+                delegate.onMessageReceivedNextMessageStream(authContext, request);
+            }
+        }
     }
 }