Blob binary transmission

pom.xml

@@ -27,7 +27,9 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.build.timestamp.format>yyyy-MM-dd HH:mm:ss</maven.build.timestamp.format>
-        <jetty.version>9.4.54.v20240208</jetty.version>
+        <jetty.version>9.4.57.v20241219</jetty.version>
+        <maven.compiler.source>1.8</maven.compiler.source>
+        <maven.compiler.target>1.8</maven.compiler.target>
     </properties>
     <build>
         <defaultGoal>install</defaultGoal>
@@ -134,6 +136,14 @@
                         </descriptorRefs>
                     </configuration>
                 </plugin>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-compiler-plugin</artifactId>
+                    <configuration>
+                        <debug>true</debug>
+                        <debuglevel>lines,vars,source</debuglevel>
+                    </configuration>
+                </plugin>
             </plugins>
         </pluginManagement>
 
@@ -149,6 +159,10 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>9</source>
+                    <target>9</target>
+                </configuration>
             </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
@@ -169,7 +183,7 @@
         <dependency>
             <groupId>net.sf.jt400</groupId>
             <artifactId>jt400</artifactId>
-            <version>11.2</version>
+            <version>21.0.5</version>
             <!-- comment the <scope> tag for local development -->
             <scope>provided</scope>
         </dependency>
@@ -213,6 +227,25 @@
             <artifactId>websocket-server</artifactId>
             <version>${jetty.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>5.11.0</version> <!-- use the latest -->
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <version>5.10.2</version> <!-- use the latest -->
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <version>5.10.2</version>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
 </project>

src/main/java/com/github/ibm/mapepire/BlobRequestData.java

@@ -0,0 +1,26 @@
+package com.github.ibm.mapepire;
+
+public class BlobRequestData {
+
+    private int replacementIndex;
+    private int length;
+    private int offset;
+
+    public BlobRequestData(int replacementIndex, int length, int offset){
+        this.replacementIndex = replacementIndex;
+        this.length = length;
+        this.offset = offset;
+    }
+
+    public int getLength() {
+        return length;
+    }
+
+    public int getReplacementIndex() {
+        return replacementIndex;
+    }
+
+    public int getOffset() {
+        return offset;
+    }
+}

src/main/java/com/github/ibm/mapepire/BlobResponseData.java

@@ -0,0 +1,34 @@
+package com.github.ibm.mapepire;
+
+import java.io.InputStream;
+import java.sql.Blob;
+
+public class BlobResponseData {
+    final Blob blob;
+    final String columnName;
+    final int rowId;
+    final int length;
+
+    public BlobResponseData(Blob is, String columnName, int rowId, int length){
+        this.blob = is;
+        this.columnName = columnName;
+        this.rowId = rowId;
+        this.length = length;
+    }
+
+    public Blob getBlob() {
+        return blob;
+    }
+
+    public String getColumnName(){
+        return columnName;
+    }
+
+    public int getRowId(){
+        return rowId;
+    }
+
+    public int getLength(){
+        return length;
+    }
+}

src/main/java/com/github/ibm/mapepire/ClientRequest.java

@@ -16,7 +16,7 @@
 public abstract class ClientRequest implements Runnable {
     private final SystemConnection m_conn;
     private final String m_id;
-    private final DataStreamProcessor m_io;
+    protected final DataStreamProcessor m_io;
     private final JsonObject m_reqObj;
     private final Map<String, Object> replyData = new LinkedHashMap<String, Object>();
 
@@ -28,10 +28,6 @@ protected ClientRequest(final DataStreamProcessor _io, final SystemConnection _c
         addReplyData("id", m_id);
     }
 
-    public SystemConnection getConnection() {
-        return m_conn;
-    }
-
     protected void addReplyData(final String _key, final Object _val) {
         replyData.put(_key, _val);
     }
@@ -126,7 +122,7 @@ private static String getErrorStringFromException(Throwable _e) {
         return "Internal Error: " + _e.getClass().getSimpleName();
     }
 
-    protected void sendreply() throws UnsupportedEncodingException, IOException {
+    protected void sendreply() throws IOException {
         final Gson l = new GsonBuilder().serializeNulls().create();
         final String json = l.toJson(replyData);
         replyData.clear();

src/main/java/com/github/ibm/mapepire/DataStreamProcessor.java

@@ -1,32 +1,37 @@
 package com.github.ibm.mapepire;
 
 import com.github.ibm.mapepire.requests.*;
+import com.github.ibm.mapepire.ws.AsyncSender;
 import com.github.theprez.jcmdutils.StringUtils;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
 
 import java.io.*;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.sql.SQLException;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.*;
 
 public class DataStreamProcessor implements Runnable {
 
     private static final Object s_replyWriterLock = new String("Response Writer Lock");
 
     private final SystemConnection m_conn;
     private final BufferedReader m_in;
-    private final PrintStream m_out;
+    private final PrintStream m_out_text;
+
     private final Map<String, RunSql> m_queriesMap = new HashMap<String, RunSql>();
     private final Map<String, PrepareSql> m_prepStmtMap = new HashMap<String, PrepareSql>();
     private final boolean m_isTestMode;
+    private final AsyncSender m_binarySender;
 
-    public DataStreamProcessor(final InputStream _in, final PrintStream _out, final SystemConnection _conn,
-            boolean _isTestMode)
+    public DataStreamProcessor(final InputStream _in, final PrintStream _outText, final AsyncSender binarySender, final SystemConnection _conn,
+                               boolean _isTestMode)
             throws UnsupportedEncodingException {
         m_in = new BufferedReader(new InputStreamReader(_in, "UTF-8"));
-        m_out = _out;
+        m_out_text = _outText;
+        m_binarySender = binarySender;
         m_conn = _conn;
         m_isTestMode = _isTestMode;
     }
@@ -66,6 +71,48 @@ public void run() {
         }
     }
 
+    private int bytesToInt(byte[] bytes, int offset, int length) {
+        int x = 0;
+        for (int i = offset; i < offset + length; i++) {
+            byte b = bytes[i];
+            x = x << 8 | b & 0xFF;
+        }
+        return x;
+    }
+
+
+    public void run(byte[] payload, int offset, int len) {
+        int cont_id_length = 2;
+        int cont_id = bytesToInt(payload, offset, cont_id_length);
+
+        List<BlobRequestData> blobRequestDataArray = new ArrayList<>();
+        int curOffset = offset + cont_id_length;
+        int replacementIndexLength = 1;
+        int sizeOfInt = 4;
+        while (curOffset < offset + len) {
+            int replacementIndex = bytesToInt(payload, curOffset, replacementIndexLength);
+            curOffset += replacementIndexLength;
+
+            int length = bytesToInt(payload, curOffset, sizeOfInt);
+            curOffset += sizeOfInt;
+
+            BlobRequestData blobRequestData = new BlobRequestData(replacementIndex, length, curOffset);
+            blobRequestDataArray.add(blobRequestData);
+            curOffset += length;
+        }
+
+        PrepareSql prev = m_prepStmtMap.get(String.valueOf(cont_id));
+        if (null == prev) {
+            dispatch(new BadReq(this, m_conn, null, "invalid correlation ID"));
+            return;
+        }
+        try {
+            RunBlob runBlob = new RunBlob(payload, blobRequestDataArray, prev);
+        } catch (Exception e) {
+            System.out.println("Caught exception " + e);
+        }
+    }
+
     public void run(String requestString) {
         final JsonElement reqElement;
         final JsonObject reqObj;
@@ -179,12 +226,85 @@ public void run(String requestString) {
 
     public void sendResponse(final String _response) throws UnsupportedEncodingException, IOException {
         synchronized (s_replyWriterLock) {
-            m_out.write((_response + "\n").getBytes("UTF-8"));
+            m_out_text.write((_response + "\n").getBytes("UTF-8"));
             Tracer.datastreamOut(_response);
-            m_out.flush();
+            m_out_text.flush();
         }
     }
 
+    public void sendResponse(final String id, final BlobResponseData blobResponseData) throws IOException, SQLException {
+        synchronized (s_replyWriterLock) {
+            int curOffset = 0;
+            byte[] buffer = new byte[4 * 1024 * 1024];
+            byte[] idBytes = id.getBytes(StandardCharsets.UTF_8);
+
+            curOffset = 0;
+            String columnName = blobResponseData.getColumnName();
+            InputStream is = blobResponseData.getBlob().getBinaryStream();
+            int rowId = blobResponseData.getRowId();
+
+            byte[] columnNameBytes = columnName.getBytes(StandardCharsets.UTF_8);
+
+            if (idBytes.length > 255) {
+                throw new IllegalArgumentException("ID too long to encode in one byte length");
+            }
+
+            // First byte is the length of the ID
+            buffer[curOffset] = (byte) idBytes.length;
+            curOffset += 1;
+
+            // Copy ID bytes after the length byte
+            System.arraycopy(idBytes, 0, buffer, curOffset, idBytes.length);
+            curOffset += idBytes.length;
+
+            // Copy rowId
+            ByteBuffer rowIdBuffer = ByteBuffer.allocate(4);
+            rowIdBuffer.putInt(rowId); // default is big-endian
+            byte[] rowIdBytes = rowIdBuffer.array();
+            for (int j = 0; j < 4; j++) {
+                buffer[curOffset] = rowIdBytes[j];
+                curOffset += 1;
+            }
+
+            // Copy column length
+            buffer[curOffset] = (byte) columnName.length();
+            curOffset += 1;
+
+            // copy column name
+            System.arraycopy(columnNameBytes, 0, buffer, curOffset, columnNameBytes.length);
+            curOffset += columnNameBytes.length;
+
+            // Copy blob length
+            ByteBuffer blobLength = ByteBuffer.allocate(4);
+            blobLength.putInt(blobResponseData.getLength()); // default is big-endian
+            byte[] bytes = blobLength.array();
+            for (int j = 0; j < 4; j++) {
+                buffer[curOffset] = bytes[j];
+                curOffset += 1;
+            }
+
+            int bytesRead = is.read(buffer, curOffset, buffer.length - curOffset);
+            curOffset += bytesRead;
+            int totalBytesRead = bytesRead;
+            if (bytesRead != -1) {
+                boolean isFinal = totalBytesRead == blobResponseData.getLength();
+                sendByteBuffer(buffer, curOffset, isFinal);
+            }
+
+            while ((bytesRead = is.read(buffer)) != -1) {
+                totalBytesRead += bytesRead;
+                boolean isFinal = totalBytesRead == blobResponseData.getLength();
+                sendByteBuffer(buffer, bytesRead, isFinal);
+            }
+        }
+    }
+
+    private void sendByteBuffer(byte[] buffer, int bytesRead, boolean isFinal) {
+        // Wrap only the bytes actually read
+        ByteBuffer byteBuffer = ByteBuffer.wrap(buffer, 0, bytesRead);
+        m_binarySender.send(byteBuffer, isFinal);
+    }
+
     public void end() {
         try {
             m_conn.getJdbcConnection().close();

src/main/java/com/github/ibm/mapepire/MapepireServer.java

@@ -10,6 +10,7 @@
 import javax.net.ssl.TrustManager;
 import javax.net.ssl.X509TrustManager;
 
+import com.github.ibm.mapepire.ws.DbWebsocketClient;
 import org.eclipse.jetty.security.ConstraintMapping;
 import org.eclipse.jetty.security.ConstraintSecurityHandler;
 import org.eclipse.jetty.server.Server;
@@ -39,6 +40,10 @@ public static boolean isSingleMode() {
         return s_isSingleMode;
     }
 
+    private static int getNumBytesInMb(int num){
+        return num * 1024 * 1024;
+    }
+
     public static void main(final String[] _args) {
 
         final LinkedList<String> args = new LinkedList<String>();
@@ -63,9 +68,8 @@ public static void main(final String[] _args) {
                 if (testMode) {
                     System.setIn(new FileInputStream(testFile));
                 }
-                final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, conn, testMode);
-
-                io.run();
+//                final DataStreamProcessor io = new DataStreamProcessor(System.in, System.out, conn, testMode);
+//                io.run();
             } else {
                 s_isSingleMode = false;
 
@@ -170,12 +174,15 @@ public static void main(final String[] _args) {
                         (servletContext, nativeWebSocketConfiguration) -> {
                             nativeWebSocketConfiguration.getPolicy().setMaxTextMessageBufferSize(65535);
                             // Configure max message size
-                            int maxWsMessageSize = 50 * 1024 * 1024; // 50MB
+                            int maxWsMessageSize = getNumBytesInMb(200);
+                            int maxBinaryMessageSize = getNumBytesInMb(200);
                             String maxWsMessageSizeStr = System.getenv("MAX_WS_MESSAGE_SIZE");
                             if (StringUtils.isNonEmpty(maxWsMessageSizeStr)) {
                                 maxWsMessageSize = Integer.parseInt(maxWsMessageSizeStr);
                             }
                             nativeWebSocketConfiguration.getPolicy().setMaxTextMessageSize(maxWsMessageSize);
+                            nativeWebSocketConfiguration.getPolicy().setMaxBinaryMessageSize(maxBinaryMessageSize);
+
 
                             // Add websockets
                             nativeWebSocketConfiguration.addMapping("/db/*", new DbSocketCreator());