Add logger support

Author: ajuvercr

gradle/libs.versions.toml

@@ -53,9 +53,6 @@ dependencies {
     implementation 'io.grpc:grpc-stub:1.63.0'
     implementation 'io.grpc:grpc-core:1.63.0'
     implementation 'com.google.protobuf:protobuf-java:4.28.2'
-    
-    // For logging, optional
-    implementation 'org.slf4j:slf4j-simple:2.0.12'
 
     implementation 'com.google.guava:guava:33.4.5-jre'
 }

runner/build.gradle

@@ -0,0 +1,84 @@
+package io.github.rdfc;
+
+import java.util.logging.Handler;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+
+import com.google.protobuf.Empty;
+
+import io.grpc.stub.StreamObserver;
+import rdfc.RunnerGrpc;
+import rdfc.Log.LogMessage;
+
+// Assuming you have a generated gRPC stub: LogServiceGrpc.LogServiceBlockingStub
+
+public class GrpcLogHandler extends Handler
+        implements StreamObserver<Empty> {
+    private final StreamObserver<LogMessage> stream;
+    private final String[] entity;
+    private final String uri;
+
+    public GrpcLogHandler(RunnerGrpc.RunnerStub stub, String uri, String... entity) {
+        this.stream = stub.logStream(this);
+        this.uri = uri;
+        this.entity = entity;
+    }
+
+    @Override
+    public void publish(LogRecord record) {
+        if (!isLoggable(record)) {
+            return;
+        }
+
+        var msg = LogMessage.newBuilder()
+                .setLevel(record.getLevel().getName().toLowerCase())
+                .setMsg(record.getMessage());
+
+        msg.addEntities(uri);
+        for (var e : entity) {
+            msg.addEntities(e);
+        }
+
+        try {
+            this.stream.onNext(msg.build()); // RPC call
+        } catch (Exception e) {
+            // Don’t let logging failures crash the app
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public void flush() {
+        // nothing to do
+    }
+
+    @Override
+    public void close() {
+        // close channel if needed
+        this.stream.onCompleted();
+    }
+
+    @Override
+    public void onNext(Empty value) {
+        throw new UnsupportedOperationException("Unimplemented method 'onNext'");
+    }
+
+    @Override
+    public void onError(Throwable t) {
+        throw new UnsupportedOperationException("Unimplemented method 'onError'");
+    }
+
+    @Override
+    public void onCompleted() {
+        throw new UnsupportedOperationException("Unimplemented method 'onCompleted'");
+    }
+
+    public static Logger createLogger(RunnerGrpc.RunnerStub stub, String uri, String... entities) {
+        var logger = Logger.getLogger(uri);
+        for (Handler h : logger.getHandlers()) {
+            logger.removeHandler(h);
+        }
+        logger.addHandler(new GrpcLogHandler(stub, uri, entities));
+        return logger;
+    }
+}

runner/src/main/java/io/github/rdfc/GrpcLogHandler.java

@@ -19,12 +19,12 @@ public static void main(String[] args) throws Exception {
         for (var arg : args) {
             System.out.println("Argument: " + arg);
         }
-        ManagedChannel channel = ManagedChannelBuilder.forTarget("dns:///127.0.0.1:50051")
+        ManagedChannel channel = ManagedChannelBuilder.forTarget(args[0])
                 .usePlaintext()
                 .build();
         RunnerGrpc.RunnerStub stub = RunnerGrpc.newStub(channel);
 
-        new Runner(stub, args[1]);
+        new Runner(stub, args[1], () -> channel.shutdown());
 
         channel.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
     }

runner/src/main/java/io/github/rdfc/Rdfc.java

@@ -44,6 +44,7 @@ public Iter<ByteString> buffers() {
     // Each listener expecting a stream, receives a stream with a single chunk
     // Each listener expecting a single message, gets that message
     void msg(ByteString buffer) {
+        System.out.println("Got msg, sending to " + this.strings.size() + " listeners");
         this.strings.forEach(string -> string.push(buffer));
         this.streams.forEach(stream -> stream.push(
                 new Reader.SingleIter<>(buffer)));

runner/src/main/java/io/github/rdfc/Reader.java

@@ -8,8 +8,11 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.*;
 import java.util.function.Consumer;
 
+import java.util.logging.*;
 import io.github.rdfc.json.ChannelHandlerModule;
 
 import com.fasterxml.jackson.databind.DeserializationFeature;
@@ -44,26 +47,49 @@ public class Runner implements StreamObserver<RunnerMessage> {
 
     protected final ObjectMapper mapper;
 
-    public Runner(RunnerGrpc.RunnerStub stub, String uri) {
+    private final AtomicInteger awaiting = new AtomicInteger(0);
+    private final Runnable onComplete;
+    private final Logger logger;
+
+    protected final String uri;
+
+    public Runner(RunnerGrpc.RunnerStub stub, String uri, Runnable onComplete) {
+        this.uri = uri;
         this.stream = stub.connect(this);
+        this.logger = GrpcLogHandler.createLogger(stub, uri, "cli");
+
+        this.onComplete = onComplete;
         this.stub = stub;
         this.mapper = new ObjectMapper();
         this.mapper.registerModule(new ChannelHandlerModule(this));
         mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 
+        this.logger.info("Hello from the runner!");
         this.sendIdentify(uri);
     }
 
     public void setReader(String uri, Reader reader) {
         this.channels.put(uri, reader);
     }
 
+    private void decreaseAndCheckEnd() {
+        var v = this.awaiting.decrementAndGet();
+        if (v == 0) {
+            this.stream.onCompleted();
+            this.onComplete.run();
+        }
+    }
+
     @Override
     public void onNext(RunnerMessage value) {
+
+        System.out.println("Got message " + value.getAllFields().keySet().toString());
+
         if (value.hasPipeline()) {
             System.out.println("Pipeline message");
             return;
         }
+
         if (value.hasMsg()) {
             System.out.println("Msg message");
             var msg = value.getMsg();
@@ -95,6 +121,7 @@ public void onNext(RunnerMessage value) {
         if (value.hasClose()) {
             System.out.println("Close message");
             var msg = value.getClose();
+
             var channel = this.channels.get(msg.getChannel());
             if (channel != null) {
                 channel.close();
@@ -107,46 +134,57 @@ public void onNext(RunnerMessage value) {
 
         if (value.hasStart()) {
             System.out.println("Start message");
+
             this.processors.forEach((k, v) -> {
                 v.produce(st -> {
                     System.out.println("Processor Produced " + k);
-                    // TODO: do something
+                    this.decreaseAndCheckEnd();
                 });
             });
+
+            System.out.println("Start happened");
             return;
         }
 
         if (value.hasProc()) {
             System.out.println("Proc start message");
             var proc = value.getProc();
             var uri = proc.getUri();
+
+            var procLogger = GrpcLogHandler.createLogger(stub, uri, this.uri);
             try {
-                Processor<?> processor = this.startProc(proc);
+                var latch = new CountDownLatch(1);
+                Processor<?> processor = this.startProc(proc, procLogger);
                 processor.init(_void -> {
+                    this.awaiting.updateAndGet(x -> x + 2);
                     processor.transform(st -> {
                         System.out.println("Processor transformed " + uri);
-                        // TODO: do something
+                        this.decreaseAndCheckEnd();
                     });
-                    this.sendProcInit(uri, Optional.empty());
+                    latch.countDown();
                 });
+                latch.await();
+
+                System.out.println("Sending proc init " + uri);
+                this.sendProcInit(uri, Optional.empty());
             } catch (Exception e) {
                 e.printStackTrace();
                 this.sendProcInit(uri, Optional.of(e.toString()));
             }
+
             return;
         }
-
         System.err.println("Unsupported message " + value.getUnknownFields());
     }
 
-    private Processor<?> startProc(rdfc.Runner.Processor proc) throws Exception {
+    private Processor<?> startProc(rdfc.Runner.Processor proc, Logger logger) throws Exception {
         var uri = proc.getUri();
         var config = proc.getConfig();
         var params = proc.getArguments();
         System.out.println("Trying to start proc " + uri + " " + config + " " + params);
 
         var arg = mapper.readValue(config, Config.class);
-        var processor = arg.loadClass(this, params);
+        var processor = arg.loadClass(this, params, logger);
 
         this.processors.put(uri, processor);
 
@@ -250,38 +288,39 @@ private static class Config {
         public String jar;
         public String clazz;
 
-        Processor<?> loadClass(Runner runner, String arguments) throws Exception {
+        Processor<?> loadClass(Runner runner, String arguments, Logger logger) throws Exception {
             URL jarUrl = new URI(this.jar).toURL();
-            try (URLClassLoader loader = new URLClassLoader(new URL[] { jarUrl }, Rdfc.class.getClassLoader())) {
-                Class<?> clazz = loader.loadClass(this.clazz);
-
-                var mapper = new ObjectMapper();
-                mapper.registerModule(new ChannelHandlerModule(runner));
-                mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
-                mapper.setTypeFactory(TypeFactory.defaultInstance().withClassLoader(loader));
-                // Find constructor with one argument
-                Constructor<?> constructor = null;
-                for (Constructor<?> c : clazz.getConstructors()) {
-                    if (c.getParameterCount() == 1) {
-                        constructor = c;
-                        break;
-                    }
-                }
-                if (constructor == null) {
-                    throw new RuntimeException("No single-arg constructor found");
+            URLClassLoader loader = new URLClassLoader(new URL[] { jarUrl }, Rdfc.class.getClassLoader());
+            Class<?> clazz = loader.loadClass(this.clazz);
+
+            var mapper = new ObjectMapper();
+            mapper.registerModule(new ChannelHandlerModule(runner));
+            mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+            mapper.setTypeFactory(TypeFactory.defaultInstance().withClassLoader(loader));
+
+            // Find constructor with one argument
+            Constructor<?> constructor = null;
+            for (Constructor<?> c : clazz.getConstructors()) {
+                if (c.getParameterCount() == 2) {
+                    constructor = c;
+                    break;
                 }
+            }
 
-                Class<?> paramType = constructor.getParameterTypes()[0];
+            if (constructor == null) {
+                throw new RuntimeException("No single-arg constructor found");
+            }
 
-                System.out.println("Found type " + paramType.getTypeName());
+            Class<?> paramType = constructor.getParameterTypes()[0];
 
-                // Use Jackson to deserialize JSON into the param type
-                Object arg = mapper.readValue(arguments, paramType);
+            System.out.println("Found type " + paramType.getTypeName());
 
-                // Instantiate using default constructor
-                constructor.setAccessible(true);
-                return (Processor<?>) constructor.newInstance(arg);
-            }
+            // Use Jackson to deserialize JSON into the param type
+            Object arg = mapper.readValue(arguments, paramType);
+
+            // Instantiate using default constructor
+            constructor.setAccessible(true);
+            return (Processor<?>) constructor.newInstance(arg, logger);
         }
     }
 }

runner/src/main/java/io/github/rdfc/Runner.java

@@ -1,2 +1,3 @@
 include 'runner'
 include 'types'
+

settings.gradle

@@ -1,16 +1,19 @@
 package io.github.rdfc;
 
 import java.util.function.Consumer;
+import java.util.logging.Logger;
 
 /**
  * Processor
  * A processor is finished when the transform and produce callbacks are called
  */
 public abstract class Processor<T> {
-    protected T arguments;
+    protected final T arguments;
+    protected final Logger logger;
 
-    public Processor(T arguments) {
+    public Processor(T arguments, Logger logger) {
         this.arguments = arguments;
+        this.logger = logger;
     }
 
     // This is called and awaits the callback before transform