Skip to content

Commit 3ea7116

Browse files
mbucktonmbuckton
committed
handle in/out for common messages and maps messages
1 parent 00a0b1e commit 3ea7116

File tree

4 files changed

+95
-61
lines changed

4 files changed

+95
-61
lines changed

‎src/main/java/io/mapsmessaging/config/protocol/impl/SatelliteConfig.java‎

Lines changed: 28 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,13 @@ public SatelliteConfig(ConfigurationProperties config) {
3737
outgoingMessagePollInterval = config.getIntProperty("outgoingMessagePollInterval", 60);
3838
httpRequestTimeout = config.getIntProperty("httpRequestTimeoutSec", 30);
3939
maxInflightEventsPerDevice = config.getIntProperty("maxInflightEventsPerDevice", 2);
40-
namespace = config.getProperty("namespaceRoot", "");
41-
inboundNamespaceRoot = config.getProperty("inboundNamespaceRoot", "/incoming/{deviceId}/{sin}/{min}");
42-
outboundNamespaceRoot = config.getProperty("outboundNamespaceRoot", "/outgoing/{deviceId}/#");
40+
4341
outboundBroadcast = config.getProperty("outboundBroadcast", "");
42+
commonInboundPublishRoot = config.getProperty("commonInboundPublishRoot", "/{deviceId}/common/in/{sin}/{min}");
43+
commonOutboundPublishRoot = config.getProperty("commonOutboundPublishRoot", "/{deviceId}/common/out/#");
44+
mapsInboundPublishRoot = config.getProperty("mapsInboundPublishRoot", "/{deviceId}/maps/in/{namespace}/#");
45+
mapsOutboundPublishRoot = config.getProperty("mapsOutboundPublishRoot", "/{deviceId}/maps/out/{namespace}/#");
46+
4447
deviceInfoUpdateMinutes = config.getIntProperty("deviceInfoUpdateMinutes", 15);
4548
maxBufferSize = config.getIntProperty("maxBufferSize", 4000);
4649
compressionCutoffSize = config.getIntProperty("compressionCutoffSize", 128);
@@ -95,10 +98,6 @@ public boolean update(BaseConfigDTO cfg) {
9598
maxInflightEventsPerDevice = dto.getMaxInflightEventsPerDevice();
9699
changed = true;
97100
}
98-
if (!outboundNamespaceRoot.equals(dto.getOutboundNamespaceRoot())) {
99-
outboundNamespaceRoot = dto.getOutboundNamespaceRoot();
100-
changed = true;
101-
}
102101
if (!outboundBroadcast.equals(dto.getOutboundBroadcast())) {
103102
outboundBroadcast = dto.getOutboundBroadcast();
104103
changed = true;
@@ -111,8 +110,23 @@ public boolean update(BaseConfigDTO cfg) {
111110
mailboxPassword = dto.getMailboxPassword();
112111
changed = true;
113112
}
114-
if(!namespace.equals(dto.getNamespace())) {
115-
namespace = dto.getNamespace();
113+
if (!commonInboundPublishRoot.equals(dto.getCommonInboundPublishRoot())) {
114+
commonInboundPublishRoot = dto.getCommonInboundPublishRoot();
115+
changed = true;
116+
}
117+
118+
if (!commonOutboundPublishRoot.equals(dto.getCommonOutboundPublishRoot())) {
119+
commonOutboundPublishRoot = dto.getCommonOutboundPublishRoot();
120+
changed = true;
121+
}
122+
123+
if (!mapsInboundPublishRoot.equals(dto.getMapsInboundPublishRoot())) {
124+
mapsInboundPublishRoot = dto.getMapsInboundPublishRoot();
125+
changed = true;
126+
}
127+
128+
if (!mapsOutboundPublishRoot.equals(dto.getMapsOutboundPublishRoot())) {
129+
mapsOutboundPublishRoot = dto.getMapsOutboundPublishRoot();
116130
changed = true;
117131
}
118132
if(deviceInfoUpdateMinutes != dto.getDeviceInfoUpdateMinutes()) {
@@ -151,18 +165,20 @@ public ConfigurationProperties toConfigurationProperties() {
151165
properties.put("outgoingMessagePollInterval", outgoingMessagePollInterval);
152166
properties.put("httpRequestTimeoutSec", httpRequestTimeout);
153167
properties.put("maxInflightEventsPerDevice", maxInflightEventsPerDevice);
154-
properties.put("outboundNamespaceRoot", outboundNamespaceRoot);
155-
properties.put("outboundBroadcast", outboundBroadcast);
156168
properties.put("mailboxId", mailboxId);
157169
properties.put("mailboxPassword", mailboxPassword);
158-
properties.put("namespaceRoot", namespace);
159170
properties.put("deviceInfoUpdateMinutes", deviceInfoUpdateMinutes);
160171
properties.put("maxBufferSize", maxBufferSize);
161172
properties.put("compressionCutoffSize", compressionCutoffSize);
162173
properties.put("messageLifeTimeInMinutes", messageLifeTimeInMinutes);
163174
properties.put("sharedSecret", sharedSecret);
164175
properties.put("sendHighPriorityMessages", sendHighPriorityMessages);
165176
properties.put("sinNumber", sinNumber);
177+
properties.put("outboundBroadcast", outboundBroadcast);
178+
properties.put("commonInboundPublishRoot", commonInboundPublishRoot);
179+
properties.put("commonOutboundPublishRoot", commonOutboundPublishRoot);
180+
properties.put("mapsInboundPublishRoot", mapsInboundPublishRoot);
181+
properties.put("mapsOutboundPublishRoot", mapsOutboundPublishRoot);
166182
return properties;
167183
}
168184
}

‎src/main/java/io/mapsmessaging/dto/rest/config/protocol/impl/OrbCommOgwsDTO.java‎

Lines changed: 0 additions & 33 deletions
This file was deleted.

‎src/main/java/io/mapsmessaging/dto/rest/config/protocol/impl/SatelliteConfigDTO.java‎

Lines changed: 36 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -39,23 +39,52 @@ public class SatelliteConfigDTO extends BaseSatelliteConfigDTO {
3939
@Schema(description = "Max number of events to be in flight per each modems")
4040
protected int maxInflightEventsPerDevice;
4141

42-
@Schema(description = "Namespace path for inbound response topic mapping for individual modems")
43-
protected String inboundNamespaceRoot;
4442

45-
@Schema(description = "Namespace path for outbound topic mapping for individual modems")
46-
protected String outboundNamespaceRoot;
43+
@Schema(
44+
description = "Topic template for publishing decoded common (SIN < 127) inbound messages (after parsing SIN/MIN).",
45+
example = "/{deviceId}/common/in/{sin}/{min}",
46+
defaultValue = "/{deviceId}/common/in/{sin}/{min}"
47+
)
48+
protected String commonInboundPublishRoot;
4749

48-
@Schema(description = "Namespace path for outbound topic broadcast for all devices")
50+
@Schema(
51+
description = "Topic root for accepting outbound common (SIN < 127) messages to be encoded and sent to the modem. Wildcards are allowed.",
52+
example = "/{deviceId}/common/out/#",
53+
defaultValue = "/{deviceId}/common/out/#"
54+
)
55+
protected String commonOutboundPublishRoot;
56+
57+
@Schema(
58+
description = "Topic template for publishing decoded MAPS (SIN 147) inbound messages into a namespace tree (after parsing).",
59+
example = "/{deviceId}/maps/in/{namespace}/#",
60+
defaultValue = "/{deviceId}/maps/in/{namespace}/#"
61+
)
62+
protected String mapsInboundPublishRoot;
63+
64+
@Schema(
65+
description = "Topic template for accepting outbound MAPS (SIN 147) messages from a namespace tree to be encoded and sent to the modem.",
66+
example = "/{deviceId}/maps/out/{namespace}/#",
67+
defaultValue = "/{deviceId}/maps/out/{namespace}/#"
68+
)
69+
protected String mapsOutboundPublishRoot;
70+
71+
@Schema(
72+
description = "Topic used to broadcast a message to all modems/clients (encoded and sent to each).",
73+
example = "/inmarsat/broadcast",
74+
defaultValue = "/inmarsat/broadcast"
75+
)
4976
protected String outboundBroadcast;
5077

78+
79+
80+
5181
@Schema(description = "Mailbox ID")
5282
protected String mailboxId;
5383

5484
@Schema(description = "Mailbox password")
5585
protected String mailboxPassword;
5686

57-
@Schema(description = "Namespace root for the mailbox to override remote MAPS message names", example="/{deviceId}", defaultValue = "")
58-
protected String namespace;
87+
5988

6089
@Schema(description = "Device Info update time in minutes")
6190
protected int deviceInfoUpdateMinutes;

‎src/main/java/io/mapsmessaging/network/protocol/impl/satellite/gateway/protocol/SatelliteGatewayProtocol.java‎

Lines changed: 31 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -68,8 +68,8 @@ public class SatelliteGatewayProtocol extends Protocol {
6868
private final AtomicReference<Map<String, List<byte[]>>> priorityMessages;
6969
private final CipherManager cipherManager;
7070

71-
private final String namespacePath;
72-
private final String incomingNamespacePath;
71+
private final String mapsIncomingNamespacePath;
72+
private final String commonIncomingNamespacePath;
7373
private final long outgoingPollInterval;
7474

7575
private final int maxBufferSize;
@@ -116,18 +116,31 @@ public SatelliteGatewayProtocol(@NonNull @NotNull EndPoint endPoint, @NotNull @N
116116
setKeepAlive(millis + random.nextLong(millis));
117117
session = SessionManager.getInstance().create(scb.build(), this);
118118
session.resumeState();
119-
String outBoundNamespacePath = config.getOutboundNamespaceRoot().trim();
119+
String outBoundNamespacePath = config.getCommonOutboundPublishRoot().trim();
120120
if(!outBoundNamespacePath.isEmpty()){
121121
String path = outBoundNamespacePath.replace("{deviceId}", primeId);
122122
path = path.replace("{mailboxId}", config.getMailboxId());
123123
SubscriptionContextBuilder subBuilder = new SubscriptionContextBuilder(path, ClientAcknowledgement.AUTO);
124124
subBuilder.setQos(QualityOfService.AT_MOST_ONCE)
125125
.setReceiveMaximum(config.getMaxInflightEventsPerDevice())
126+
.setAlias("common_requests")
126127
.setNoLocalMessages(true);
127128
session.addSubscription(subBuilder.build());
128129
}
129-
namespacePath = parsePath(config.getNamespace(), "", primeId, config.getMailboxId());
130-
incomingNamespacePath = parsePath(config.getInboundNamespaceRoot(), "/{deviceId}/incoming/{sin}/{min}", primeId, config.getMailboxId());
130+
String mapsOutboundNamespacePath = config.getMapsOutboundPublishRoot();
131+
if(!mapsOutboundNamespacePath.isEmpty()){
132+
String path = mapsOutboundNamespacePath.replace("{deviceId}", primeId);
133+
path = path.replace("{mailboxId}", config.getMailboxId());
134+
SubscriptionContextBuilder subBuilder = new SubscriptionContextBuilder(path, ClientAcknowledgement.AUTO);
135+
subBuilder.setQos(QualityOfService.AT_MOST_ONCE)
136+
.setReceiveMaximum(config.getMaxInflightEventsPerDevice())
137+
.setAlias("maps_requests")
138+
.setNoLocalMessages(true);
139+
session.addSubscription(subBuilder.build());
140+
}
141+
142+
mapsIncomingNamespacePath = parsePath(config.getMapsInboundPublishRoot(), "/{deviceId}/maps/in", primeId, config.getMailboxId());
143+
commonIncomingNamespacePath = parsePath(config.getCommonInboundPublishRoot(), "/{deviceId}/common/in/{sin}/{min}", primeId, config.getMailboxId());
131144
String bcast = config.getOutboundBroadcast();
132145
if(bcast != null && !bcast.isEmpty()){
133146
SubscriptionContextBuilder subBuilder = new SubscriptionContextBuilder(bcast, ClientAcknowledgement.AUTO);
@@ -200,7 +213,8 @@ public ProtocolInformationDTO getInformation() {
200213

201214
@Override
202215
public void sendMessage(@NotNull @NonNull MessageEvent messageEvent) {
203-
if (messageEvent.getDestinationName().endsWith("request")) {
216+
if(messageEvent.getSubscription().getContext().getAlias() != null &&
217+
messageEvent.getSubscription().getContext().getAlias().equals("common_requests")){
204218
MessageData messageData = new MessageData();
205219
byte[] tmp = messageEvent.getMessage().getOpaqueData();
206220
int sin = tmp[0] & 0xFF;
@@ -348,7 +362,7 @@ public void handleIncomingMessage(MessageData message) throws ExecutionException
348362
private void handleCommonMessage(MessageData message, byte[] raw) throws ExecutionException, InterruptedException {
349363
int sin = message.getSin() & 0xff;
350364
int min = message.getMin() & 0xff;
351-
String path = incomingNamespacePath;
365+
String path = commonIncomingNamespacePath;
352366
path = path.replace("{sin}", String.valueOf(sin));
353367
path = path.replace("{min}", String.valueOf(min));
354368
logger.log(SATELLITE_RECEIVED_RAW_MESSAGE, sin, min, raw.length, path);
@@ -369,10 +383,18 @@ private void processMapsMessage(SatelliteMessage satelliteMessage, Map<String, S
369383
destinationCount++;
370384
messageCount += entry.getValue().size();
371385
String topic = entry.getKey();
372-
if(namespacePath != null && !namespacePath.isEmpty()){
373-
topic = namespacePath + "/" + topic;
386+
boolean isSchema = false;
387+
if(topic.toLowerCase().startsWith("$schema")){
388+
topic = topic.substring("$schema".length());
389+
isSchema = true;
390+
}
391+
if(mapsIncomingNamespacePath != null && !mapsIncomingNamespacePath.isEmpty()){
392+
topic = mapsIncomingNamespacePath + "/" + topic;
374393
topic = topic.replace("//", "/"); // to be sure
375394
}
395+
if(isSchema){
396+
topic = "$schema"+topic;
397+
}
376398
publishEvents(topic, entry.getValue(), transformation1, meta);
377399
}
378400
logger.log(SATELLITE_RECEIVED_PACKED_MESSAGE, destinationCount, messageCount, satelliteMessage.getMessage().length, raw.length);

0 commit comments

Comments
 (0)