Implement blocking generic serial transport for java

examples/java/example/apps/CustomTransport.java

@@ -0,0 +1,51 @@
+package example.apps;
+
+import zcm.zcm.*;
+import javazcm.types.example_t;
+import java.io.IOException;
+
+public class CustomTransport implements ZCMSubscriber
+{
+    public void messageReceived(ZCM zcm, String channel, ZCMDataInputStream ins)
+    {
+        try {
+            example_t msg = new example_t(ins);
+            System.out.println("msg timestamp: " + msg.timestamp);
+        } catch (IOException ex) {
+            System.out.println("Exception: " + ex);
+        }
+    }
+
+    public static void main(String[] args) throws IOException
+    {
+        CustomTransport t = new CustomTransport();
+
+        // Create a loopback SerialIO implementation for testing
+        LoopbackSerialIO serialIO = new LoopbackSerialIO();
+
+        // Create the ZCM transport using the SerialIO
+        ZCMTransport transport = new ZCMGenericSerialTransport(serialIO, 1 << 14, 1 << 17);
+        ZCM zcm = new ZCM(transport);
+
+        String channel = "EXAMPLE";
+
+        zcm.subscribe(channel, t);
+        zcm.start();
+
+        example_t msg = new example_t();
+        msg.position = new double[] { 1, 2, 3 };
+        msg.orientation = new double[] { 1, 0, 0, 0 };
+        msg.ranges = new short[] {
+            0, 1, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14
+        };
+        msg.num_ranges = msg.ranges.length;
+        msg.name = "example string";
+        msg.enabled = true;
+
+        while (true) {
+            msg.timestamp = System.currentTimeMillis() * 1000;
+            zcm.publish(channel, msg);
+            try { Thread.sleep(100); } catch (InterruptedException ex) {}
+        }
+    }
+}

examples/java/example/apps/LoopbackSerialIO.java

@@ -0,0 +1,159 @@
+package example.apps;
+
+import zcm.zcm.SerialIO;
+import java.nio.ByteBuffer;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A loopback SerialIO implementation for testing purposes.
+ * This implementation uses internal queues to simulate serial communication,
+ * where data written to the "serial port" can be read back from it.
+ * This is useful for testing ZCM serial transport without actual hardware.
+ */
+public class LoopbackSerialIO implements SerialIO {
+
+    private final BlockingQueue<Byte> dataQueue;
+    private final int maxQueueSize;
+
+    /**
+     * Create a new loopback SerialIO with default queue size.
+     */
+    public LoopbackSerialIO() {
+        this(10000); // Default queue size of 10KB
+    }
+
+    /**
+     * Create a new loopback SerialIO with specified queue size.
+     *
+     * @param maxQueueSize maximum number of bytes to buffer internally
+     */
+    public LoopbackSerialIO(int maxQueueSize) {
+        this.maxQueueSize = maxQueueSize;
+        this.dataQueue = new LinkedBlockingQueue<>(maxQueueSize);
+    }
+
+    /**
+     * Read data from the internal queue (simulating serial read).
+     *
+     * @param buffer ByteBuffer to store the read data
+     * @param maxLen maximum number of bytes to read
+     * @param timeoutMs timeout in milliseconds, 0 for non-blocking
+     * @return number of bytes actually read
+     */
+    @Override
+    public int get(ByteBuffer buffer, int maxLen, int timeoutMs) {
+        if (buffer == null || maxLen <= 0) {
+            return 0;
+        }
+
+        int bytesRead = 0;
+
+        if (timeoutMs == 0) {
+            // Non-blocking mode - drain available bytes directly into buffer
+            while (bytesRead < maxLen && !dataQueue.isEmpty()) {
+                Byte b = dataQueue.poll();
+                if (b != null) {
+                    buffer.put(b);
+                    bytesRead++;
+                }
+            }
+        } else {
+            // Blocking mode - wait for at least one byte, then drain available
+            try {
+                Byte firstByte = dataQueue.poll(timeoutMs, TimeUnit.MILLISECONDS);
+                if (firstByte != null) {
+                    buffer.put(firstByte);
+                    bytesRead = 1;
+                    // Drain any additional available bytes (non-blocking)
+                    while (bytesRead < maxLen && !dataQueue.isEmpty()) {
+                        Byte b = dataQueue.poll();
+                        if (b != null) {
+                            buffer.put(b);
+                            bytesRead++;
+                        }
+                    }
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return 0;
+            }
+        }
+
+        return bytesRead;
+    }
+
+    /**
+     * Write data to the internal queue (simulating serial write).
+     *
+     * @param buffer ByteBuffer containing data to write
+     * @param len number of bytes to write from the buffer
+     * @param timeoutMs timeout in milliseconds, 0 for non-blocking
+     * @return number of bytes actually written
+     */
+    @Override
+    public int put(ByteBuffer buffer, int len, int timeoutMs) {
+        if (buffer == null || len <= 0) {
+            return 0;
+        }
+
+        int bytesWritten = 0;
+        int bytesToWrite = Math.min(len, buffer.remaining());
+
+        for (int i = 0; i < bytesToWrite; i++) {
+            byte b = buffer.get();
+            boolean success = false;
+
+            if (timeoutMs == 0) {
+                // Non-blocking mode
+                success = dataQueue.offer(b);
+            } else {
+                // Blocking mode with timeout
+                try {
+                    success = dataQueue.offer(b, timeoutMs, TimeUnit.MILLISECONDS);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    // Put the byte back to the buffer since it wasn't written
+                    buffer.position(buffer.position() - 1);
+                    break;
+                }
+            }
+
+            if (success) {
+                bytesWritten++;
+            } else {
+                // Queue is full or timeout, put the byte back and break
+                buffer.position(buffer.position() - 1);
+                break;
+            }
+        }
+
+        return bytesWritten;
+    }
+
+    /**
+     * Get the current number of bytes available for reading.
+     *
+     * @return number of bytes in the internal queue
+     */
+    public int available() {
+        return dataQueue.size();
+    }
+
+    /**
+     * Clear all data from the internal queue.
+     */
+    public void clear() {
+        dataQueue.clear();
+    }
+
+    /**
+     * Get the maximum queue size.
+     *
+     * @return maximum number of bytes this SerialIO can buffer
+     */
+    public int getMaxQueueSize() {
+        return maxQueueSize;
+    }
+}

examples/java/example/apps/SerialIOTest.java

@@ -0,0 +1,96 @@
+package example.apps;
+
+import zcm.zcm.SerialIO;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * A simple test runner to demonstrate and validate SerialIO implementations
+ * without requiring the full ZCM infrastructure.
+ */
+public class SerialIOTest {
+
+    public static void main(String[] args) {
+        System.out.println("ZCM SerialIO Test Suite");
+        System.out.println("=======================");
+
+        // Test LoopbackSerialIO
+        testLoopbackSerialIO();
+
+        System.out.println("\nAll tests completed!");
+    }
+
+    /**
+     * Test the LoopbackSerialIO implementation
+     */
+    private static void testLoopbackSerialIO() {
+        System.out.println("\n--- Testing LoopbackSerialIO ---");
+
+        LoopbackSerialIO serialIO = new LoopbackSerialIO(1000);
+
+        // Test data
+        String testMessage = "Hello ZCM Serial!";
+        byte[] testData = testMessage.getBytes(StandardCharsets.UTF_8);
+
+        // Test writing
+        ByteBuffer writeBuffer = ByteBuffer.allocateDirect(testData.length);
+        writeBuffer.put(testData);
+        writeBuffer.flip();
+
+        int bytesWritten = serialIO.put(writeBuffer, testData.length, 0);
+        System.out.println("Wrote " + bytesWritten + " bytes: \"" + testMessage + "\"");
+
+        // Test reading
+        ByteBuffer readBuffer = ByteBuffer.allocateDirect(1000);
+        int bytesRead = serialIO.get(readBuffer, 1000, 0);
+
+        if (bytesRead > 0) {
+            readBuffer.flip();
+            byte[] readData = new byte[bytesRead];
+            readBuffer.get(readData);
+            String receivedMessage = new String(readData, StandardCharsets.UTF_8);
+            System.out.println("Read " + bytesRead + " bytes: \"" + receivedMessage + "\"");
+
+            if (testMessage.equals(receivedMessage)) {
+                System.out.println("LoopbackSerialIO test PASSED");
+            } else {
+                System.out.println("LoopbackSerialIO test FAILED - data mismatch");
+            }
+        } else {
+            System.out.println("LoopbackSerialIO test FAILED - no data read");
+        }
+
+        // Test queue status
+        System.out.println("Available bytes: " + serialIO.available());
+        System.out.println("Max queue size: " + serialIO.getMaxQueueSize());
+    }
+
+    /**
+     * Utility method to demonstrate ByteBuffer operations
+     */
+    private static void demonstrateByteBufferUsage() {
+        System.out.println("\n--- ByteBuffer Usage Demo ---");
+
+        // Create a direct ByteBuffer (recommended for SerialIO)
+        ByteBuffer buffer = ByteBuffer.allocateDirect(100);
+        System.out.println("Created direct buffer with capacity: " + buffer.capacity());
+
+        // Write some data
+        String data = "Sample data";
+        buffer.put(data.getBytes(StandardCharsets.UTF_8));
+        System.out.println("After writing, position: " + buffer.position() + ", remaining: " + buffer.remaining());
+
+        // Flip buffer for reading
+        buffer.flip();
+        System.out.println("After flip, position: " + buffer.position() + ", limit: " + buffer.limit());
+
+        // Read the data back
+        byte[] readData = new byte[buffer.remaining()];
+        buffer.get(readData);
+        System.out.println("Read back: \"" + new String(readData, StandardCharsets.UTF_8) + "\"");
+
+        // Clear buffer for reuse
+        buffer.clear();
+        System.out.println("After clear, position: " + buffer.position() + ", limit: " + buffer.limit());
+    }
+}