bdware/cm
2
0
Fork 0
mirror of https://gitee.com/BDWare/cm synced 2025-01-10 01:44:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: update event mechanism

Browse Source 
add second centers for event topics
This commit is contained in:
Frank.R.Wu 2022-01-20 20:55:49 +08:00
parent 1ec1cac894
commit bb368c8bee
2 changed files with 28 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/bdware/sc/event/EventBroker.java
View File

@ -106,7 +106,7 @@ public class EventBroker {
doSubscribe(event); doSubscribe(event);
// save & try to sub in center // save & try to sub in center
recorder.appendEvent(event); recorder.appendEvent(event);
center.subInCenter(event.getTopic(), event.getSemantics()); center.subInCenter(event.getTopic(), event.getSemantics(), event.getCenter());
} }
break; break;
case UNSUBSCRIBE: case UNSUBSCRIBE:

37
src/main/java/org/bdware/sc/event/EventCenter.java
View File

@ -15,17 +15,18 @@ public class EventCenter {
* get the nearest node to the topic in the hash function range * get the nearest node to the topic in the hash function range
* *
* @param topic the topic * @param topic the topic
* @param k the number of centers
* @return id of the node * @return id of the node
*/ */
public String getCenterByTopic(String topic) { public String[] getCenterByTopic(String topic, int k) {
if (null == instance.nodeCenterConn) { if (null == instance.nodeCenterConn) {
return null; return null;
} }
String[] centers = DHTUtil.getClusterByKey(topic, 1); String[] centers = DHTUtil.getClusterByKey(topic, k);
if (null == centers || centers.length == 0) { if (null == centers || centers.length == 0) {
return null; return null;
} }
return centers[0]; return centers;
} }
/** /**
@ -33,21 +34,34 @@ public class EventCenter {
* *
* @param topic event topic * @param topic event topic
* @param semantics event semantics, used to mark PRESUB events * @param semantics event semantics, used to mark PRESUB events
* @param center id of event center if the subscribing has been handled
*/ */
public void subInCenter(String topic, REvent.REventSemantics semantics) { public void subInCenter(String topic, REvent.REventSemantics semantics, String center) {
if (null == instance.nodeCenterConn) { if (null == instance.nodeCenterConn) {
return; return;
} }
REvent msg = new REvent( REvent msg = new REvent(topic,
topic,
REvent.REventType.SUBSCRIBE, REvent.REventType.SUBSCRIBE,
String.format("{\"subscriber\":\"%s\"}", String.format("{\"subscriber\":\"%s\"}",
instance.nodeCenterConn.getNodeId()), instance.nodeCenterConn.getNodeId()),
""); "");
msg.setSemantics(semantics); msg.setSemantics(semantics);
msg.doSignature(instance.nodeCenterConn.getNodeKeyPair()); msg.doSignature(instance.nodeCenterConn.getNodeKeyPair());
String nodeId = getCenterByTopic(topic); msg.setCenter(center);
instance.masterStub.deliverEvent(nodeId, JsonUtil.toJson(msg)); String[] centers = getCenterByTopic(topic, 2);
if (null == centers) {
return;
}
if (null != center) {
if (centers[0].equals(center) && centers.length > 1) {
instance.masterStub.deliverEvent(centers[1], JsonUtil.toJson(msg));
} else {
instance.masterStub.deliverEvent(centers[0], JsonUtil.toJson(msg));
}
} else if (!instance.masterStub.deliverEvent(centers[0], JsonUtil.toJson(msg)) && centers.length > 1) {
msg.setCenter(centers[0]);
instance.masterStub.deliverEvent(centers[1], JsonUtil.toJson(msg));
}
} }
/** /**
@ -61,8 +75,11 @@ public class EventCenter {
if (null == instance.masterStub) { if (null == instance.masterStub) {
return false; return false;
} }
String nodeId = getCenterByTopic(topic); String[] centers = getCenterByTopic(topic, 1);
return instance.masterStub.deliverEvent(nodeId, JsonUtil.toJson(event)); if (null == centers) {
return false;
}
return instance.masterStub.deliverEvent(centers[0], JsonUtil.toJson(event));
} }
/** /**