+ public ClusterWrapper getClusterWrapper() {
+ return clusterWrapper;
+ }
+
+ public String getCurrentMemberName(){
+ return clusterWrapper.getCurrentMemberName();
+ }
+
+ /**
+ * Send the message to each and every shard
+ *
+ * @param message
+ */
+ public void broadcast(Object message){
+ for(String shardName : configuration.getAllShardNames()){
+
+ Optional<ActorSelection> primary = findPrimaryShard(shardName);
+ if (primary.isPresent()) {
+ primary.get().tell(message, ActorRef.noSender());
+ } else {
+ LOG.warn("broadcast failed to send message {} to shard {}. Primary not found",
+ message.getClass().getSimpleName(), shardName);
+ }
+ }
+ }
+
+ public FiniteDuration getOperationDuration() {
+ return operationDuration;
+ }
+
+ public boolean isPathLocal(String path) {
+ if (Strings.isNullOrEmpty(path)) {
+ return false;
+ }
+
+ int pathAtIndex = path.indexOf('@');
+ if (pathAtIndex == -1) {
+ //if the path is of local format, then its local and is co-located
+ return true;
+
+ } else if (selfAddressHostPort != null) {
+ // self-address and tx actor path, both are of remote path format
+ int slashIndex = path.indexOf('/', pathAtIndex);
+
+ if (slashIndex == -1) {
+ return false;
+ }
+
+ String hostPort = path.substring(pathAtIndex + 1, slashIndex);
+ return hostPort.equals(selfAddressHostPort);
+
+ } else {
+ // self address is local format and tx actor path is remote format
+ return false;
+ }
+ }
+