Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

VMRouter should add message tracking information #6368

Draft
wants to merge 4 commits into
base: development
Choose a base branch
from
Draft

VMRouter should add message tracking information #6368

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5446c71
initial enhancement
rogin Nov 28, 2024
fa544a8
remove unnecessary method
rogin Dec 1, 2024
bc03aeb
refactor to a single class
rogin Dec 2, 2024
5bdcf9d
rename function
rogin Dec 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions server/conf/mirth.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,3 +112,6 @@ database.connection.retrywaitinmilliseconds = 10000
# database-readonly.url = jdbc:...
#
database.enable-read-write-split = true

# If true, enable routing enhancements for improved message tracking
routing.enable-enhancements = true
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why should this be a configurable option? What harmful behavior could happen if the routing was done for a use case that did not expect it?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had in mind any existing scripts that are looking for specific data that may break.

282 changes: 248 additions & 34 deletions server/src/com/mirth/connect/server/userutil/VMRouter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,14 @@

package com.mirth.connect.server.userutil;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

Expand All @@ -28,77 +36,117 @@ public class VMRouter {
private ChannelController channelController = ControllerFactory.getFactory().createChannelController();
private EngineController engineController = ControllerFactory.getFactory().createEngineController();

private TrackingEnhancer trackingEnhancer;

/**
* Instantiates a VMRouter object.
*/
public VMRouter() {}

/**
* Instantiates a VMRouter object with additional message tracking enhancements.
*
* @param channelId channel ID or "NONE" if null
* @param messageId message ID or -1L if null
* @param sourceMap the message's source map
*/
public VMRouter(String channelId, Long messageId, SourceMap sourceMap) {
this.trackingEnhancer = new TrackingEnhancer(channelId, messageId, sourceMap);
}

/**
* Dispatches a message to a channel, specified by the deployed channel name. If the dispatch
* fails for any reason (for example, if the target channel is not started), a Response object
* with the ERROR status and the error message will be returned.
* fails for any reason (for example, if the target channel is not started), a {@link Response} object
* with the {@link Status#ERROR} status and the error message will be returned.
*
* @param channelName
* The name of the deployed channel to dispatch the message to.
* @param message
* The message to dispatch to the channel.
* @return The Response object returned by the channel, if its source connector is configured to
* @param channelName The name of the deployed channel to dispatch the message to.
* @param message The message to dispatch to the channel.
* @return The {@link Response} object returned by the channel, if its source connector is configured to
* return one.
*/
public Response routeMessage(String channelName, String message) {
return routeMessage(channelName, new RawMessage(message));
return routeMessage(channelName, createRawMessage(message, null, null));
}

/**
* Dispatches a message to a channel, specified by the deployed channel name. If the dispatch
* fails for any reason (for example, if the target channel is not started), a Response object
* with the ERROR status and the error message will be returned.
* fails for any reason (for example, if the target channel is not started), a {@link Response} object
* with the {@link Status#ERROR} status and the error message will be returned.
*
* @param channelName
* The name of the deployed channel to dispatch the message to.
* @param rawMessage
* A RawMessage object to dispatch to the channel.
* @return The Response object returned by the channel, if its source connector is configured to
* @param channelName The name of the deployed channel to dispatch the message to.
* @param rawMessage A {@link RawMessage} object to dispatch to the channel.
* @return The {@link Response} object returned by the channel, if its source connector is configured to
* return one.
*/
public Response routeMessage(String channelName, RawMessage rawMessage) {
com.mirth.connect.model.Channel channel = channelController.getDeployedChannelByName(channelName);

if (channel == null) {
logger.error("Could not find channel to route to for channel name: " + channelName);
return new Response(Status.ERROR, "Could not find channel to route to for channel name: " + channelName);
String message = "Could not find channel to route to for channel name: " + channelName;
logger.error(message);
return new Response(Status.ERROR, message);
}

return routeMessageByChannelId(channel.getId(), rawMessage);
}

/**
Route a message to the specified channelName. Information about the chain of source channel Ids and
source message Ids will be included in the sourceMap of the downstream message automatically in a
similar manner as if a Channel Writer was being used.

@param channelName The name of the channel to which to route the message.
@param message The content of the message to be sent, textual or binary. As String or byte[].
@param sourceMap A map containing entries to include in the sourceMap of the sent message.
@return The {@link Response} object returned by the channel.

@see #routeMessage(String, Object, Map, Collection)
*/
public Response routeMessage(String channelName, Object message, Map<String, Object> sourceMap) {
return routeMessage(channelName, message, sourceMap, null);
}

/**
Route a message to the specified channelName. Information about the chain of source channel Ids and
source message Ids will be included in the sourceMap of the downstream message automatically in a
similar manner as if a Channel Writer was being used.

@param channelName The name of the channel to which to route the message.
@param message The content of the message to be sent, textual or binary. As String or byte[].
@param sourceMap A map containing entries to include in the sourceMap of the sent message.
@param destinationSet A collection of integers (metadata IDs) representing which destinations to dispatch the message to.
Null may be passed to indicate all destinations. If unspecified, all destinations is the default.
@return The {@link Response} object returned by the channel.

@see VMRouter#routeMessage(String, RawMessage)
*/
public Response routeMessage(String channelName, Object message, Map<String, Object> sourceMap, Collection<Number> destinationSet) {
return routeMessage(channelName, createRawMessage(message, sourceMap, destinationSet));
}

/**
* Dispatches a message to a channel, specified by the deployed channel ID. If the dispatch
* fails for any reason (for example, if the target channel is not started), a Response object
* with the ERROR status and the error message will be returned.
* fails for any reason (for example, if the target channel is not started), a {@link Response} object
* with the {@link Status#ERROR} status and the error message will be returned.
*
* @param channelId
* The ID of the deployed channel to dispatch the message to.
* @param message
* The message to dispatch to the channel.
* @return The Response object returned by the channel, if its source connector is configured to
* return one.
* @param channelId The ID of the deployed channel to dispatch the message to.
* @param message The message to dispatch to the channel.
* @return The {@link Response} object returned by the channel, if its source connector is configured to
* return one.
*/
public Response routeMessageByChannelId(String channelId, String message) {
return routeMessageByChannelId(channelId, new RawMessage(message));
return routeMessageByChannelId(channelId, createRawMessage(message, null, null));
}

/**
* Dispatches a message to a channel, specified by the deployed channel ID. If the dispatch
* fails for any reason (for example, if the target channel is not started), a Response object
* with the ERROR status and the error message will be returned.
* fails for any reason (for example, if the target channel is not started), a {@link Response} object
* with the {@link Status#ERROR} status and the error message will be returned.
*
* @param channelId
* The ID of the deployed channel to dispatch the message to.
* @param rawMessage
* A RawMessage object to dispatch to the channel.
* @return The Response object returned by the channel, if its source connector is configured to
* return one.
* @param channelId The ID of the deployed channel to dispatch the message to.
* @param rawMessage A {@link RawMessage} object to dispatch to the channel.
* @return The {@link Response} object returned by the channel, if its source connector is configured to
* return one.
*/
public Response routeMessageByChannelId(String channelId, RawMessage rawMessage) {
try {
Expand All @@ -119,11 +167,177 @@ public Response routeMessageByChannelId(String channelId, RawMessage rawMessage)
}
}

/**
Route a message to the specified channelId. Information about the chain of source channel Ids and
source message Ids will be included in the sourceMap of the downstream message automatically in a
similar manner as if a Channel Writer was being used.

@param channelId The unique identifier of the channel to which to route the message.
@param message The content of the message to be sent, textual or binary. As String or byte[].
@return The {@link Response} object returned by the channel.

@see #routeMessageByChannelId(String, Object, Map, Collection)
*/
public Response routeMessageByChannelId(String channelId, Object message) {
return routeMessageByChannelId(channelId, message, null, null);
}

/**
Route a message to the specified channelId. Information about the chain of source channel Ids and
source message Ids will be included in the sourceMap of the downstream message automatically in a
similar manner as if a Channel Writer was being used.

@param channelId The unique identifier of the channel to which to route the message.
@param message The content of the message to be sent, textual or binary. As String or byte[].
@param sourceMap A map containing entries to include in the sourceMap of the sent message.
@return The {@link Response} object returned by the channel.

@see #routeMessageByChannelId(String, Object, Map, Collection)
*/
public Response routeMessageByChannelId(String channelId, Object message, Map<String, Object> sourceMap) {
return routeMessageByChannelId(channelId, message, sourceMap, null);
}

/**
Route a message to the specified channelId. Information about the chain of source channel Ids and
source message Ids will be included in the sourceMap of the downstream message automatically in a
similar manner as if a Channel Writer was being used.

@param channelId The unique identifier of the channel to which to route the message.
@param message The content of the message to be sent, textual or binary. As String or byte[].
@param sourceMap A map containing entries to include in the sourceMap of the sent message.
@param destinationSet A collection of integers (metadata IDs) representing which destinations to dispatch the message to.
Null may be passed to indicate all destinations. If unspecified, all destinations is the default.
@return The {@link Response} object returned by the channel.

@see {@link VMRouter#routeMessageByChannelId(String, RawMessage)}
*/
public Response routeMessageByChannelId(String channelId, Object message, Map<String, Object> sourceMap, Collection<Number> destinationSet) {
return routeMessageByChannelId(channelId, createRawMessage(message, sourceMap, destinationSet));
}

private com.mirth.connect.donkey.model.message.RawMessage convertRawMessage(RawMessage message) {
if (message.isBinary()) {
return new com.mirth.connect.donkey.model.message.RawMessage(message.getRawBytes(), message.getDestinationMetaDataIds(), message.getSourceMap());
} else {
return new com.mirth.connect.donkey.model.message.RawMessage(message.getRawData(), message.getDestinationMetaDataIds(), message.getSourceMap());
}
}

/**
Create a {@link RawMessage} with the specified content, sourceMap, and destinationSet.

@param message The content of the message to be sent, textual or binary. As String or byte[].
@param sourceMap A map containing entries to include in the sourceMap of the {@link RawMessage} (optional).
@param destinationSet A collection of integers (metadata IDs) representing which destinations to dispatch the message to.
Null may be passed to indicate all destinations. If unspecified, all destinations is the default (optional).
@return A {@link RawMessage} object containing the message, source, and destination information.
*/
public RawMessage createRawMessage(Object message, Map<String, Object> sourceMap, Collection<Number> destinationSet) {
if(trackingEnhancer != null) {
sourceMap = trackingEnhancer.enrich(sourceMap);
}

if(message instanceof byte[]) {
return new RawMessage((byte[])message, destinationSet, sourceMap);
} else {
return new RawMessage(message.toString(), destinationSet, sourceMap);
}
}

/**
* Adds additional message tracking data.
*
* TrackingEnhancer
*/
private class TrackingEnhancer {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there any value in making this a not inner class?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps for testing. Can certainly do so if folks find it beneficial.

private String channelId;
private Long messageId;
private SourceMap envSourceMap;

/**
* Constructor
*
* @param channelId channel ID; null defaults to "NONE"
* @param messageId message ID; null defaults to -1L
* @param sourceMap the message's source map
*/
private TrackingEnhancer(String channelId, Long messageId, SourceMap sourceMap) {
this.channelId = channelId != null ? channelId : "NONE";
this.messageId = messageId != null ? messageId : -1L;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is this -1L and not zero?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I followed the existing template logic which is a reasonable "non-value" value.

this.envSourceMap = sourceMap;
}

/**
* Enrich the given source map with additional message tracking data.
*
* @param messageSourceMap
* @return a new Map
*/
private Map<String, Object> enrich(Map<String, Object> messageSourceMap) {
if (messageSourceMap == null) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Leaving the condition is fine but is there any chance of messageSourceMap ever being null?

Should the @Null or @NotNull annotations be used?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I see it can come in as null from at least one method.

I don't see other places where @Null and @NotNull are used that I could copy, so I defer to others on properly decorating and validating the code.

messageSourceMap = Collections.emptyMap();
}

List<String> sourceChannelIds = envLookupAsList("sourceChannelIds", "sourceChannelId");
List<String> sourceMessageIds = envLookupAsList("sourceMessageIds", "sourceMessageId");

HashMap<String,Object> newSourceMap = new HashMap<String,Object>(messageSourceMap);
String channelId = this.channelId;
Long messageId = this.messageId;

sourceChannelIds.add(channelId);
sourceMessageIds.add(messageId.toString());

newSourceMap.put("sourceChannelIds", sourceChannelIds);
newSourceMap.put("sourceChannelId", channelId);
newSourceMap.put("sourceMessageIds", sourceMessageIds);
newSourceMap.put("sourceMessageId", messageId);

return newSourceMap;
}

/**
* Given the specified lookup keys, return the first non-null value as a List.
* The expectation is the first lookup will return a List, while the second returns an Object.
*
* @param primary primary lookup key to return a List
* @param secondary secondary lookup key to return an Object
* @return a List containing the first non-null lookup value, else an empty List
*/
private List<String> envLookupAsList(String primary, String secondary) {
rogin marked this conversation as resolved.
Show resolved Hide resolved
List<String> result = new ArrayList<String>();

Object primaryValue = lookupInEnvSourceMap(primary);

if(primaryValue != null) {
//all of this to not assume the result is a List<String>
if(primaryValue instanceof Collection) {
((Collection<?>)primaryValue).stream()
.map(i -> i.toString())
.forEach(result::add);
} else if(primaryValue instanceof Object[]) {
Arrays.stream((Object[])primaryValue)
.map(i -> i.toString())
.forEach(result::add);
}
} else {
Object secondaryValue = lookupInEnvSourceMap(secondary);
if(secondaryValue != null) {
result.add(secondaryValue.toString());
}
}

return result;
}

/**
* Look up a value from the environment's {@link SourceMap}
* @param key
* @return its mapped value, can be null
*/
private Object lookupInEnvSourceMap(String key) {
return this.envSourceMap.get(key);
}
}
}
Loading