You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by dl...@apache.org on 2021/04/14 22:20:00 UTC
[asterixdb] 02/07: [NO ISSUE][IPC] +=
sendRealTimeMessageTo[Primary]CC to INCMessageBroker
This is an automated email from the ASF dual-hosted git repository.
dlych pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit d58a5ace7800cad5231ddfaad2338ff886c4dff0
Author: Michael Blow <mb...@apache.org>
AuthorDate: Mon Apr 12 22:18:00 2021 -0400
[NO ISSUE][IPC] += sendRealTimeMessageTo[Primary]CC to INCMessageBroker
Change-Id: I0d8a8e2018a1839f20a7c1b601c26c8a3502ef5d
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/11026
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Till Westmann <ti...@apache.org>
---
.../apache/asterix/messaging/NCMessageBroker.java | 10 +++++++++
.../common/messaging/api/INCMessageBroker.java | 24 ++++++++++++++++++----
2 files changed, 30 insertions(+), 4 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
index 26f6524..1b216ea 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
@@ -79,6 +79,16 @@ public class NCMessageBroker implements INCMessageBroker {
}
@Override
+ public void sendRealTimeMessageToCC(CcId ccId, ICcAddressedMessage message) throws Exception {
+ ncs.sendRealTimeApplicationMessageToCC(ccId, JavaSerializationUtils.serialize(message), null);
+ }
+
+ @Override
+ public void sendRealTimeMessageToPrimaryCC(ICcAddressedMessage message) throws Exception {
+ sendRealTimeMessageToCC(ncs.getPrimaryCcId(), message);
+ }
+
+ @Override
public void sendMessageToNC(String nodeId, INcAddressedMessage message) throws Exception {
IChannelControlBlock messagingChannel = ncs.getMessagingNetworkManager().getMessagingChannel(nodeId);
sendMessageToChannel(messagingChannel, message);
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java
index 6bd58a9..88905fd 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/messaging/api/INCMessageBroker.java
@@ -29,7 +29,7 @@ public interface INCMessageBroker extends IMessageBroker {
* @param message
* @throws Exception
*/
- public void sendMessageToPrimaryCC(ICcAddressedMessage message) throws Exception;
+ void sendMessageToPrimaryCC(ICcAddressedMessage message) throws Exception;
/**
* Sends application message from this NC to the CC.
@@ -37,7 +37,23 @@ public interface INCMessageBroker extends IMessageBroker {
* @param message
* @throws Exception
*/
- public void sendMessageToCC(CcId ccId, ICcAddressedMessage message) throws Exception;
+ void sendMessageToCC(CcId ccId, ICcAddressedMessage message) throws Exception;
+
+ /**
+ * Sends high-priority application message from this NC to the primary CC.
+ *
+ * @param message
+ * @throws Exception
+ */
+ void sendRealTimeMessageToPrimaryCC(ICcAddressedMessage message) throws Exception;
+
+ /**
+ * Sends high-priority application message from this NC to the CC.
+ *
+ * @param message
+ * @throws Exception
+ */
+ void sendRealTimeMessageToCC(CcId ccId, ICcAddressedMessage message) throws Exception;
/**
* Sends application message from this NC to another NC.
@@ -45,14 +61,14 @@ public interface INCMessageBroker extends IMessageBroker {
* @param message
* @throws Exception
*/
- public void sendMessageToNC(String nodeId, INcAddressedMessage message) throws Exception;
+ void sendMessageToNC(String nodeId, INcAddressedMessage message) throws Exception;
/**
* Queue a message to this {@link INCMessageBroker} for processing
*
* @param msg
*/
- public void queueReceivedMessage(INcAddressedMessage msg);
+ void queueReceivedMessage(INcAddressedMessage msg);
/**
* Creates and registers a Future for a message that will be send through this broker