Use sockets

camel/pom.xml

@@ -60,6 +60,10 @@
       <artifactId>camel-stream</artifactId>
       <version>3.14.7</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.camel</groupId>
+      <artifactId>camel-netty</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.camel</groupId>
       <artifactId>camel-jdbc</artifactId>

camel/src/main/java/com/github/theprez/manzan/ManzanMessageFormatter.java

@@ -20,6 +20,10 @@ public static String format(final String _in, final Map<String, Object> _mapping
         return formatter.format(_mappings);
     }
 
+    public String getM_fmtStr() {
+        return m_fmtStr;
+    }
+
     private final String m_fmtStr;
 
     public ManzanMessageFormatter(final String _fmtStr) {
@@ -33,8 +37,9 @@ public String format(final Map<String, Object> _mappings) {
         ret = ret.replace("\\n", "\n").replace("\\t", "\t");
 
         for (final Entry<String, Object> repl : _mappings.entrySet()) {
-            ret = ret.replace("$" + repl.getKey() + "$", "" + repl.getValue());
-            String jsonIndicator ="$json:" + repl.getKey() + "$"; 
+            final String key = repl.getKey();
+            ret = ret.replace("$" + key + "$", "" + repl.getValue());
+            String jsonIndicator ="$json:" + key + "$";
             if(ret.contains(jsonIndicator)) {
                 ret = ret.replace(jsonIndicator, jsonEncode("" + repl.getValue()));
             }

camel/src/main/java/com/github/theprez/manzan/routes/event/WatchMsgEvent.java

@@ -2,6 +2,9 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.model.dataformat.JsonLibrary;
 
 import com.github.theprez.jcmdutils.StringUtils;
 import com.github.theprez.manzan.ManzanEventType;
@@ -15,29 +18,31 @@ public class WatchMsgEvent extends ManzanRoute {
     private final String m_schema;
     private final String m_sessionId;
     private final ManzanMessageFormatter m_formatter;
+    private final String m_socketIp = "0.0.0.0";
+    private final String m_socketPort = "8080";
+
 
-    public WatchMsgEvent(final String _name, final String _session_id, final String _format, final List<String> _destinations, final String _schema, final int _interval, final int _numToProcess) throws IOException {
+    public WatchMsgEvent(final String _name, final String _session_id, final String _format,
+            final List<String> _destinations, final String _schema, final int _interval, final int _numToProcess)
+            throws IOException {
         super(_name);
         m_interval = _interval;
         m_numToProcess = _numToProcess;
         m_schema = _schema;
-        m_formatter = StringUtils.isEmpty(_format) ? null: new ManzanMessageFormatter(_format);
+        m_formatter = StringUtils.isEmpty(_format) ? null : new ManzanMessageFormatter(_format);
         super.setRecipientList(_destinations);
         m_sessionId = _session_id.trim().toUpperCase();
     }
 
 //@formatter:off
     @Override
     public void configure() {
-        from("timer://foo?synchronous=true&period=" + m_interval)
-        .routeId("manzan_msg:"+m_name)
-        .setHeader(EVENT_TYPE, constant(ManzanEventType.WATCH_MSG))
-        .setBody(constant("SeLeCt * fRoM " + m_schema + ".mAnZaNmSg wHeRe SESSION_ID = '"+m_sessionId+"' limit " + m_numToProcess ))
-       // .to("stream:out")
-        .to("jdbc:jt400?outputType=StreamList")
-        .split(body()).streaming().parallelProcessing()
-            .setHeader("id", simple("${body[ORDINAL_POSITION]}"))
-            .setHeader("session_id", simple("${body[SESSION_ID]}"))
+        from(String.format("netty:tcp://%s:%s?sync=false", m_socketIp, m_socketPort))
+            .log("Received raw message: ${body}")
+            .unmarshal().json(JsonLibrary.Jackson, Map.class)
+            .routeId("manzan_msg:"+m_name)
+            .setHeader(EVENT_TYPE, constant(ManzanEventType.WATCH_MSG))
+            .setHeader("session_id", simple("${body[sessionId]}"))
             .setHeader("data_map", simple("${body}"))
             .marshal().json(true) //TODO: skip this if we are applying a format
             .setBody(simple("${body}\n"))
@@ -46,9 +51,7 @@ public void configure() {
                     exchange.getIn().setBody(m_formatter.format(getDataMap(exchange)));
                 }
             })
-            .recipientList(constant(getRecipientList())).parallelProcessing().stopOnException().end()
-            .setBody(simple("delete fRoM " + m_schema + ".mAnZaNmSg where ORDINAL_POSITION = ${header.id} WITH NC"))
-            .to("jdbc:jt400").to("stream:err");
+            .recipientList(constant(getRecipientList())).parallelProcessing().stopOnException().end();
     }
     //@formatter:on
 

ile/Makefile

@@ -19,16 +19,16 @@ src/mzversion.h:
 /qsys.lib/${BUILDLIB}.lib:
 	system "RUNSQL SQL('create schema ${BUILDLIB}')  COMMIT(*NONE) NAMING(*SQL) "
 
-/qsys.lib/${BUILDLIB}.lib/handler.pgm: /qsys.lib/${BUILDLIB}.lib/handler.module /qsys.lib/${BUILDLIB}.lib/pub_json.module /qsys.lib/${BUILDLIB}.lib/pub_db2.module /qsys.lib/${BUILDLIB}.lib/debug.module /qsys.lib/${BUILDLIB}.lib/pub_db2.module /qsys.lib/${BUILDLIB}.lib/userconf.module
+/qsys.lib/${BUILDLIB}.lib/handler.pgm: /qsys.lib/${BUILDLIB}.lib/handler.module /qsys.lib/${BUILDLIB}.lib/pub_json.module /qsys.lib/${BUILDLIB}.lib/pub_db2.module /qsys.lib/${BUILDLIB}.lib/debug.module /qsys.lib/${BUILDLIB}.lib/pub_db2.module /qsys.lib/${BUILDLIB}.lib/userconf.module /qsys.lib/${BUILDLIB}.lib/SockClient.module
 
 /qsys.lib/${BUILDLIB}.lib/%.pgm:
 	system "CRTPGM PGM(${BUILDLIB}/$*) MODULE($(patsubst %.module,$(BUILDLIB)/%,$(notdir $^))) ACTGRP(*CALLER)"
 
 /qsys.lib/${BUILDLIB}.lib/%.module: src/%.cpp src/mzversion.h
-	system "CRTCPPMOD MODULE(${BUILDLIB}/$*) SRCSTMF('$(CURDIR)/$<') OPTION(*EVENTF) SYSIFCOPT(*IFS64IO) DBGVIEW(*SOURCE) TERASPACE(*YES *TSIFC) STGMDL(*SNGLVL) DTAMDL(*p128) DEFINE(DEBUG_ENABLED)" 
+	system "CRTCPPMOD MODULE(${BUILDLIB}/$*) SRCSTMF('$(CURDIR)/$<') OPTION(*EVENTF) SYSIFCOPT(*IFS64IO) DBGVIEW(*SOURCE) TERASPACE(*YES *TSIFC) STGMDL(*SNGLVL) DTAMDL(*p128) DEFINE(DEBUG_ENABLED) OUTPUT(*PRINT) TGTCCSID(*JOB)" 
 
 /qsys.lib/${BUILDLIB}.lib/%.module: src/%.sqlc
-	system "CRTSQLCI OBJ(${BUILDLIB}/$*) SRCSTMF('$(CURDIR)/$^') COMMIT(*NONE) DATFMT(*ISO) TIMFMT(*ISO) DBGVIEW(*SOURCE) CVTCCSID(*JOB) COMPILEOPT('INCDIR(''src'')') SQLPATH(${BUILDLIB}) DFTRDBCOL(${BUILDLIB}) OPTION(*SQL)" 
+	system "CRTSQLCPPI OBJ(${BUILDLIB}/$*) SRCSTMF('$(CURDIR)/$^') COMMIT(*NONE) DATFMT(*ISO) TIMFMT(*ISO) DBGVIEW(*SOURCE) CVTCCSID(*JOB) COMPILEOPT('INCDIR(''src'')') SQLPATH(${BUILDLIB}) DFTRDBCOL(${BUILDLIB}) OPTION(*SQL)" 
 
 /qsys.lib/${BUILDLIB.lib}:
 	-system "RUNSQL SQL('create schema ${BUILDLIB}') NAMING(*SYS)"

ile/src/SockClient.cpp

@@ -0,0 +1,69 @@
+#include <iostream>
+#include <cstring>  // For memset and memcpy
+#include <sys/socket.h>  // For socket functions
+#include <arpa/inet.h>  // For inet_addr
+#include <unistd.h>  // For close
+#include "SockClient.h"
+#include "manzan.h"
+
+// Constructor to initialize the socket descriptor to -1
+SockClient::SockClient(){
+    sock_fd = -1;
+} 
+
+// Method to open a socket and connect to the server
+bool SockClient::openSocket(const std::string ip, int port) {
+    // Create socket
+    sock_fd = socket(AF_INET, SOCK_STREAM, 0);
+    if (sock_fd < 0) {
+        DEBUG_ERROR("Error creating socket\n");
+        return false;
+    }
+
+    // Define server address
+    struct sockaddr_in server_address;
+    server_address.sin_family = AF_INET;
+    server_address.sin_port = htons(port);
+    server_address.sin_addr.s_addr = inet_addr(const_cast<char*>(ip.c_str()));
+
+    // Connect to server
+    if (connect(sock_fd, (struct sockaddr*)&server_address, sizeof(server_address)) < 0) {
+        DEBUG_ERROR("Error connecting to server\n");
+        closeSocket();
+        return false;
+    }
+
+    DEBUG_INFO("Connected to server at %s:%d\n", ip.c_str(), port); 
+    return true;
+}
+
+// Method to send a message (string) over the socket
+bool SockClient::sendMessage(const std::string message) {
+    if (sock_fd < 0) {
+        DEBUG_ERROR("Socket is not open\n");
+        return false;
+    }
+
+    int bytes_sent = send(sock_fd, const_cast<char*>(message.c_str()), message.size(), 0);
+    if (bytes_sent < 0) {
+        DEBUG_ERROR("Error sending message\n");
+        return false;
+    }
+    DEBUG_INFO("Sent message: %s\n", message);
+
+    return true;
+}
+
+// Method to close the socket
+void SockClient::closeSocket() {
+    if (sock_fd >= 0) {
+        close(sock_fd);
+        sock_fd = -1;
+        DEBUG_INFO("Socket closed\n");
+    }
+}
+
+// Destructor to ensure socket is closed
+SockClient::~SockClient() {
+    closeSocket();
+}

ile/src/SockClient.h

@@ -0,0 +1,23 @@
+#include <string>
+
+// extern "C" {
+class SockClient {
+
+int sock_fd; 
+public:
+    // Constructor to initialize the socket descriptor to -1
+    SockClient();
+
+    // Method to open a socket and connect to the server
+    bool openSocket(const std::string ip, int port);
+
+    // Method to send a message (string) over the socket
+    bool sendMessage(const std::string message);
+
+    // Method to close the socket
+    void closeSocket();
+
+    // Destructor to ensure socket is closed
+    ~SockClient();
+};
+// }

ile/src/handler.cpp

@@ -12,6 +12,8 @@
 #include "event_data.h"
 #include "userconf.h"
 #include "mzversion.h"
+#include "pub_json.h"
+#include "SockClient.h"
 
 static FILE *fd = NULL;