-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
15 changed files
with
417 additions
and
55 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
47 changes: 47 additions & 0 deletions
47
camel/src/main/java/com/github/theprez/manzan/routes/event/WatchMsgEventSockets.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| package com.github.theprez.manzan.routes.event; | ||
|
|
||
| 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; | ||
| import com.github.theprez.manzan.ManzanMessageFormatter; | ||
| import com.github.theprez.manzan.routes.ManzanRoute; | ||
|
|
||
| public class WatchMsgEventSockets extends ManzanRoute { | ||
|
|
||
| private final ManzanMessageFormatter m_formatter; | ||
| private final String m_socketIp = "0.0.0.0"; | ||
| private final String m_socketPort = "8080"; | ||
|
|
||
| public WatchMsgEventSockets(final String _name, final String _format, | ||
| final List<String> _destinations, final String _schema, final int _interval, final int _numToProcess) | ||
| throws IOException { | ||
| super(_name); | ||
| m_formatter = StringUtils.isEmpty(_format) ? null : new ManzanMessageFormatter(_format); | ||
| super.setRecipientList(_destinations); | ||
| } | ||
|
|
||
| //@formatter:off | ||
| @Override | ||
| public void configure() { | ||
| from(String.format("netty:tcp://%s:%s?sync=false", m_socketIp, m_socketPort)) | ||
| .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")) | ||
| .process(exchange -> { | ||
| if (null != m_formatter) { | ||
| exchange.getIn().setBody(m_formatter.format(getDataMap(exchange))); | ||
| } | ||
| }) | ||
| .recipientList(constant(getRecipientList())).parallelProcessing().stopOnException().end(); | ||
| } | ||
| //@formatter:on | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| ## Messaging | ||
|
|
||
| ### Messaging Preference | ||
|
|
||
| By default, Manzan will send messages between the Handler component and the Camel component via socket communication. If however socket communication happens to fail for any reason, it will use SQL based communication as a fallback option. If instead, you prefer Manzan to use SQL based communication as a first option, you can set the environment variable `MANZAN_MESSAGING_PREFERENCE=SQL`. Set the environment variable back to the value `SOCKETS` to set your preference to socket communication. | ||
|
|
||
| ### Messaging options | ||
| Manzan supports two options for sending messages between the Handler and the Camel component (SQL and socket communication).\ | ||
| **SOCKET COMMUNICATION:** This option provides real time communication and is faster than the sql option.\ | ||
| **SQL COMMUNICATION:** Using this option works via the Handler component inserting data into a Db2 table, and then the Camel component subsequently reading from the table. This option isn't quite as fast as socket communication, however the data is more durable in the case that the Camel component is malfunctioning. | ||
|
|
||
| Socket communication is recommended, because by providing a fallback option to SQL in the case of socket communication malfunctioning, we guarantee data will not be lost. | ||
|
|
||
| * `MANZAN_MESSAGING_PREFERENCE = SQL`: Prefer SQL based communication | ||
| * `MANZAN_MESSAGING_PREFERENCE != SQL`: Prefer socket based communication | ||
|
|
||
| ### Setting the MANZAN_MESSAGING_PREFERENCE | ||
|
|
||
| The `MANZAN_MESSAGING_PREFERENCE` environment variable can be set with the `ADDENVVAR` command. For example, to set the messaging preference to prefer SQL based communication run the command: | ||
|
|
||
| ```cl | ||
| ADDENVVAR ENVVAR(MANZAN_MESSAGING_PREFERENCE) VALUE(SQL) LEVEL(*SYS) REPLACE(*YES) | ||
| ``` | ||
|
|
||
| After which you will need to run `ENDTCPSVR *SSHD` and `STRTCPSVR *SSHD`. Then restart your SSH session for the new environment variable to take effect. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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(); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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(); | ||
| }; | ||
| // } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.