Java 类com.hazelcast.core.Message 实例源码

项目：scalable-websocket-chat-with-hazelcast 文件：ChatMessageEventListener.java

@Override
public void onMessage(Message<ChatMessage> event) {
    System.out.println("HZ ChatMessage Topic Listener invoked");
    ChatMessage msg = event.getMessageObject();

    Predicate<Session> filterCriteria = null;
    if (!msg.isPrivate()) {
        //for ALL (except self)
        filterCriteria = (session) -> !session.getUserProperties().get("user").equals(msg.from());

    } else {
        String privateRecepient = msg.getRecepient();
        //private IM
        filterCriteria = (session) -> privateRecepient.equals(session.getUserProperties().get("user"));
    }

    ChatServer.getSessions().stream()
            .filter(filterCriteria)
            .forEach((session) -> session.getAsyncRemote().sendObject(new Reply(msg.getMsg(), msg.from(), msg.isPrivate())));
}

项目：beyondj 文件：HazelcastSessionDiscoveryEventListener.java

@Override
public void onMessage(Message<SessionDataDTO> sessionDataDTOMessage) {
    SessionDataDTO sessionDataDTO = sessionDataDTOMessage.getMessageObject();

    try {
        String thisServer = IpAddresssUtil.getLocalHostAddress();
        for (SessionDiscoveryCallback callback : callBacks) {
            String location = sessionDataDTO.getLocation();
            if (thisServer.equals(location)) {
                if (LOG.isDebugEnabled()) LOG.debug("Ignoring session discovery message as it originated on this server");
                continue;
            } else {
                if (LOG.isDebugEnabled()) LOG.debug("Updating with session discovery message on this server");
                callback.update(sessionDataDTO);
            }
        }
    } catch (Exception e) {
        LOG.error("Failed to act on message", e);
    }
}

项目：reactive-data 文件：DistributedPipedInputStream.java

@Override
public void onMessage(Message<byte[]> message) {
  if(closed)
    return;

  if(message.getPublishingMember().localMember())
  {
    log.debug("Ignoring bytes received from self..");
    return;
  }
  byte[] bytesRecvd = message.getMessageObject();
  log.debug("Got bytes of length- "+bytesRecvd.length);
  try
  {
    handleBytesReceived(bytesRecvd);
  }
  finally
  {

  }


}

项目：reactive-data 文件：FileChunkReceiver.java

@Override
public void onMessage(final Message<FileChunk> message) {
  if(message.getPublishingMember().localMember())
  {
    return;
  }
  if(isMarkDiscard()){
    //log.debug(message.getMessageObject()+"");
    discardChunks++;
    return;
  }
  try 
  {
    boolean offered = queue.offer(message.getMessageObject(), 10, TimeUnit.SECONDS);
    if(!offered)
      log.error("Chunk lost. Unable to queue even after waiting 10 secs!");

  } catch (InterruptedException e) {
    Thread.currentThread().interrupt();

  }

}

项目：hazelcast-hibernate5 文件：TimestampsRegionCacheTest.java

@Test
public void shouldUseClusterTimestampFromInvalidationmessageInsteadOfSystemTime() {
    long firstTimestamp = 1;
    long secondTimestamp = 2;
    long publishTime = 3;
    long clusterTime = 4;

    when(cluster.getClusterTime()).thenReturn(firstTimestamp, secondTimestamp);

    // cache is primed by call, that uses clusterTime instead of system clock for timestamp
    assertThat(target.put("QuerySpace", firstTimestamp, firstTimestamp, null), is(true));

    assertThat("primed value should be in the cache", (Long)target.get("QuerySpace", firstTimestamp), is(firstTimestamp));

    // a message is generated on a different cluster member informing us to update the timestamp region cache
    Message<Object> message = new Message<Object>("topicName", new Timestamp("QuerySpace", secondTimestamp), publishTime, member);

    // process the timestamp region update
    listener.onMessage(message);

    // this fails if we use system time instead of cluster time, causing the value to stay invisible until clustertime == system time (which it often isn't)
    assertThat("key should be visible and have value specified in timestamp message, with current cluster time.", (Long)target.get("QuerySpace", clusterTime), is(secondTimestamp));
}

项目：hazelcast-hibernate 文件：TimestampsRegionCacheTest.java

@Test
public void shouldUseClusterTimestampFromInvalidationmessageInsteadOfSystemTime() {
    long firstTimestamp = 1;
    long secondTimestamp = 2;
    long publishTime = 3;
    long clusterTime = 4;

    when(cluster.getClusterTime()).thenReturn(firstTimestamp, secondTimestamp);

    // cache is primed by call, that uses clusterTime instead of system clock for timestamp
    assertThat(target.put("QuerySpace", firstTimestamp, firstTimestamp, null), is(true));

    assertThat("primed value should be in the cache", (Long)target.get("QuerySpace", firstTimestamp), is(firstTimestamp));

    // a message is generated on a different cluster member informing us to update the timestamp region cache
    Message<Object> message = new Message<Object>("topicName", new Timestamp("QuerySpace", secondTimestamp), publishTime, member);

    // process the timestamp region update
    listener.onMessage(message);

    // this fails if we use system time instead of cluster time, causing the value to stay invisible until clustertime == system time (which it often isn't)
    assertThat("key should be visible and have value specified in timestamp message, with current cluster time.", (Long)target.get("QuerySpace", clusterTime), is(secondTimestamp));
}

项目：hazelcast-hibernate 文件：TimestampsRegionCacheTest.java

@Test
public void shouldUseClusterTimestampFromInvalidationmessageInsteadOfSystemTime() {
    long firstTimestamp = 1;
    long secondTimestamp = 2;
    long publishTime = 3;
    long clusterTime = 4;

    when(cluster.getClusterTime()).thenReturn(firstTimestamp, secondTimestamp);

    // cache is primed by call, that uses clusterTime instead of system clock for timestamp
    assertThat(target.put("QuerySpace", firstTimestamp, firstTimestamp, null), is(true));

    assertThat("primed value should be in the cache", (Long)target.get("QuerySpace", firstTimestamp), is(firstTimestamp));

    // a message is generated on a different cluster member informing us to update the timestamp region cache
    Message<Object> message = new Message<Object>("topicName", new Timestamp("QuerySpace", secondTimestamp), publishTime, member);

    // process the timestamp region update
    listener.onMessage(message);

    // this fails if we use system time instead of cluster time, causing the value to stay invisible until clustertime == system time (which it often isn't)
    assertThat("key should be visible and have value specified in timestamp message, with current cluster time.", (Long)target.get("QuerySpace", clusterTime), is(secondTimestamp));
}

项目：OpenCyclos 文件：HazelcastCacheManagerPeerProvider.java

@Override
public void onMessage(final Message<CacheEvictionEvent> message) {
    CacheEvictionEvent event = message.getMessageObject();
    if (LOG.isDebugEnabled()) {
        LOG.debug("Received cache eviction event: " + event);
    }
    Ehcache cache = cacheManager.getEhcache(event.getName());
    // No need to evict something in a cache which was not yet created
    if (cache != null) {
        Object key = event.getKey();
        if (key == null) {
            cache.removeAll(true);
        } else {
            cache.remove(key, true);
        }
    }
}

项目：hybridbpm 文件：DashBoardNotificationMessageListener.java

@Override
public void onMessage(final Message<DashboardNotificationEvent> message) {
    try {
        hybridbpmUI.access(new Runnable() {
            @Override
            public void run() {
                DashboardNotificationEvent event = message.getMessageObject();
                switch (event.getAction()) {
                    case SHOW:
                        hybridbpmUI.getMainMenu().changeNotification(event.getViewUrl(), true, event.getMessage());
                        break;
                    case REMOVE:
                        hybridbpmUI.getMainMenu().changeNotification(event.getViewUrl(), false, null);
                        break;
                }

                hybridbpmUI.push();
            }
        });
    } catch (Exception ex) {
        logger.log(Level.SEVERE, ex.getMessage(), ex);
    }
}

项目：DSC 文件：HazelcastService.java

@Override
public void progressFlowMessage(Message<FlowMessage> Message) {//处理流表订阅消息
    FlowMessage flowMessage = Message.getMessageObject();
    Map<String,Object> rawValues = new HashMap<String,Object>();

    String json = flowMessage.json;
    System.out.println(json);
    try {
        rawValues = FlowEntryPushUtil.jsonToStorageEntry(json);
        int state = FlowEntryPushUtil.checkFlow(rawValues);
        if(state == 0 ){
            storageSourceService.insertRowAsync(StaticFlowEntryPusher.TABLE_NAME, rawValues);
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

项目：NoMQ 文件：NoMQMessageListener.java

@Override
public void onMessage(final Message<Event> eventMessage) {
    lockTemplate.lock(() -> {
        final Event event = eventMessage.getMessageObject();
        if (synced) {
            log.debug("Recording event [id={}]", event.id());
            recordEventStore.append(eventMessage.getMessageObject());
            if (!isSyncRequest(event)) {
                playbackQueue.add(event);
            }
        } else {
            if (!isSyncRequest(event)) {
                tempPlaybackQueue.add(event);
            }
        }
    });
}

项目：hazelcast-archive 文件：TopicMBean.java

@SuppressWarnings("unchecked")
@Override
public void postRegister(Boolean registrationDone) {
    super.postRegister(registrationDone);
    if (!registrationDone) {
        return;
    }
    if (managementService.showDetails()) {
        servedStats = ManagementService.newStatisticsCollector();
        listener = new MessageListener() {

            public void onMessage(Message msg) {
                servedStats.addEvent();
            }
        };
        getManagedObject().addMessageListener(listener);
    }
}

项目：hazelcast-archive 文件：HazelcastClientTopicTest.java

@Test
public void addMessageListener() throws InterruptedException {
    HazelcastClient hClient = getHazelcastClient();
    ITopic<String> topic = hClient.getTopic("addMessageListener");
    final CountDownLatch latch = new CountDownLatch(1);
    final String message = "Hazelcast Rocks!";
    topic.addMessageListener(new MessageListener<String>() {
        public void onMessage(Message<String> msg) {
            if (msg.getMessageObject().equals(message)) {
                latch.countDown();
            }
        }
    });
    topic.publish(message);
    assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
}

项目：hazelcast-archive 文件：HazelcastClientTopicTest.java

@Test
    public void removeMessageListener() throws InterruptedException {
        HazelcastClient hClient = getHazelcastClient();
        ITopic<String> topic = hClient.getTopic("removeMessageListener");
        final CountDownLatch latch = new CountDownLatch(2);
        final CountDownLatch cp = new CountDownLatch(1);
//        final String message = "Hazelcast Rocks!";
        MessageListener<String> messageListener = new MessageListener<String>() {
            public void onMessage(Message<String> msg) {
//                if (msg.startsWith(message)) {
                System.out.println("Received " + msg + " at " + this);
                latch.countDown();
                cp.countDown();
//                }
            }
        };
        final String message = "message_" + messageListener.hashCode() + "_";
        topic.addMessageListener(messageListener);
        topic.publish(message + "1");
        cp.await();
        topic.removeMessageListener(messageListener);
        topic.publish(message + "2");
        Thread.sleep(50);
        assertEquals(1, latch.getCount());
    }

项目：hazelcast-archive 文件：MessageListenerManagerTest.java

@Test
public void testRemoveMessageListener() throws Exception {
    MessageListenerManager manager = new MessageListenerManager();
    String name = "default";
    assertTrue(manager.noListenerRegistered(name));
    MessageListener listener = new MessageListener<Object>() {

        public void onMessage(Message<Object> message) {
        }
    };
    manager.registerListener(name, listener);
    assertFalse(manager.noListenerRegistered(name));
    manager.removeListener(name, listener);
    assertTrue(manager.noListenerRegistered(name));
    manager.removeListener(name, listener);
    assertTrue(manager.noListenerRegistered(name));
}

项目：NeverwinterDP-Commons 文件：HazelcastClusterService.java

public void onMessage(Message<ClusterEvent> message) {
  long start = System.currentTimeMillis() ;
  ClusterEvent event = message.getMessageObject() ;
  Timer.Context timeCtx = appMonitor.timer("event", event.getType().toString()).time() ;
  logger.info("Start onMessage(...), event = " + event.getType());
  for(int i = 0; i < listeners.size(); i++) {
    ClusterListener<Server> listener = listeners.get(i) ;
    listener.onEvent(server, event) ;
  }
  long end = System.currentTimeMillis() ;
  String msg = "Received an event " +  event.getType() + " " + event.getSource() + " from " + event.getSourceMember().toString();
  String activityLogName = event.getType().toString() ;
  timeCtx.stop() ;
  ActivityLog log = new ActivityLog(activityLogName, ActivityLog.Type.ClusterEvent, start, end, msg) ;
  server.getActivityLogs().add(log);
  logger.info(log.toString());
  logger.info("Finish onMessage(...), event = " + event.getType());
}

项目：andes 文件：HZBasedDatabaseSyncNotificationListener.java

/**
 * {@inheritDoc}
 */
@Override
public void onMessage(Message message) {
    //TODO: do we need to handle this only for non-local members?

    ClusterNotification clusterNotification = (ClusterNotification) message.getMessageObject();
    if (log.isDebugEnabled()) {
        log.debug("Handling cluster gossip: received a DB sync request " + clusterNotification
                .getEncodedObjectAsString());
    }
    log.info("DB sync request received after a split brain recovery from cluster " + message
            .getPublishingMember());
    for (DBSyncNotificationHandler dbSyncNotificationListener : dbSyncNotificationListeners) {
        dbSyncNotificationListener.handleClusterNotification(clusterNotification);
    }
    log.info("DB sync completed for the request from cluster " + message.getPublishingMember());

}

项目：moquette 文件：HazelcastListener.java

@Override
public void onMessage(Message<HazelcastMsg> msg) {
    ByteBuf payload = null;
    try {
        if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) {
            HazelcastMsg hzMsg = msg.getMessageObject();
            LOG.info("{} received from hazelcast for topic {} message: {}", hzMsg.getClientId(), hzMsg.getTopic(),
                hzMsg.getPayload());
            // TODO pass forward this information in somehow publishMessage.setLocal(false);

            MqttQoS qos = MqttQoS.valueOf(hzMsg.getQos());
            MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, false, 0);
            MqttPublishVariableHeader varHeader = new MqttPublishVariableHeader(hzMsg.getTopic(), 0);
            payload = Unpooled.wrappedBuffer(hzMsg.getPayload());
            MqttPublishMessage publishMessage = new MqttPublishMessage(fixedHeader, varHeader, payload);
            server.internalPublish(publishMessage, hzMsg.getClientId());
        }
    } catch (Exception ex) {
        LOG.error("error polling hazelcast msg queue", ex);
    } finally {
        ReferenceCountUtil.release(payload);
    }
}

项目：hazelcast-simulator 文件：ReliableTopicTest.java

@Override
public void onMessage(Message<MessageEntity> message) {
    String threadId = message.getMessageObject().thread;
    Long previousValue = values.get(threadId);
    if (previousValue == null) {
        previousValue = 0L;
    }

    long actualValue = message.getMessageObject().value;
    long expectedValue = previousValue + 1;
    if (expectedValue != actualValue) {
        failures.incrementAndGet();
        ExceptionReporter.report(testContext.getTestId(), new TestException(format(
                "There is an unexpected gap or equality between values. Expected %d, but was %d",
                expectedValue, actualValue)));
    }

    values.put(threadId, actualValue);

    if (received.getAndIncrement() % 100000 == 0) {
        logger.info(toString() + " is at " + message.getMessageObject().toString());
    }
}

项目：open-cyclos 文件：HazelcastCacheManagerPeerProvider.java

@Override
public void onMessage(final Message<CacheEvictionEvent> message) {
    CacheEvictionEvent event = message.getMessageObject();
    if (LOG.isDebugEnabled()) {
        LOG.debug("Received cache eviction event: " + event);
    }
    Ehcache cache = cacheManager.getEhcache(event.getName());
    // No need to evict something in a cache which was not yet created
    if (cache != null) {
        Object key = event.getKey();
        if (key == null) {
            cache.removeAll(true);
        } else {
            cache.remove(key, true);
        }
    }
}

项目：mule-module-publish-subscribe 文件：PubSubModule.java

/**
 * Subscribe for Mule events under the specified topic name
 * <p/>
 * {@sample.xml ../../../doc/pubsub-module.xml.sample pubsub:listener}
 *
 * @param topic    Name of the topic
 * @param callback flow to process
 */
@Source(exchangePattern = MessageExchangePattern.ONE_WAY)
public void listener(String topic, final SourceCallback callback) {
    ITopic hazelcastTopic = HazelcastManager.getInstance().getHazelcastInstance().getTopic(topic);
    hazelcastTopic.addMessageListener(new MessageListener() {
        @Override
        public void onMessage(Message message) {
            Thread.currentThread().setContextClassLoader(muleContext.getExecutionClassLoader());

            MuleEvent newEvent = createMuleEvent(message);

            // process it
            try {
                callback.processEvent(newEvent);
            } catch (MuleException e) {
                LOGGER.error(e.getMessage(), e);
            }
        }
    });
}

项目：health-and-care-developer-network 文件：LocalRegionCache.java

protected MessageListener<Object> createMessageListener() {
    return new MessageListener<Object>() {
        public void onMessage(final Message<Object> message) {
            final Invalidation invalidation = (Invalidation) message.getMessageObject();
            if (versionComparator != null) {
                final Value value = cache.get(invalidation.getKey());
                if (value != null) {
                    Object currentVersion = value.getVersion();
                    Object newVersion = invalidation.getVersion();
                    if (versionComparator.compare(newVersion, currentVersion) > 0) {
                        cache.remove(invalidation.getKey(), value);
                    }
                }
            } else {
                cache.remove(invalidation.getKey());
            }
        }
    };
}

项目：health-and-care-developer-network 文件：TopicMBean.java

@SuppressWarnings("unchecked")
@Override
public void postRegister(Boolean registrationDone) {
    super.postRegister(registrationDone);
    if (!registrationDone) {
        return;
    }
    if (managementService.showDetails()) {
        servedStats = ManagementService.newStatisticsCollector();
        listener = new MessageListener() {

            public void onMessage(Message msg) {
                servedStats.addEvent();
            }
        };
        getManagedObject().addMessageListener(listener);
    }
}

项目：health-and-care-developer-network 文件：HazelcastClientTopicTest.java

@Test
public void addMessageListener() throws InterruptedException {
    HazelcastClient hClient = getHazelcastClient();
    ITopic<String> topic = hClient.getTopic("addMessageListener");
    final CountDownLatch latch = new CountDownLatch(1);
    final String message = "Hazelcast Rocks!";
    topic.addMessageListener(new MessageListener<String>() {
        public void onMessage(Message<String> msg) {
            if (msg.getMessageObject().equals(message)) {
                latch.countDown();
            }
        }
    });
    topic.publish(message);
    assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
}

项目：health-and-care-developer-network 文件：HazelcastClientTopicTest.java

@Test
    public void removeMessageListener() throws InterruptedException {
        HazelcastClient hClient = getHazelcastClient();
        ITopic<String> topic = hClient.getTopic("removeMessageListener");
        final CountDownLatch latch = new CountDownLatch(2);
        final CountDownLatch cp = new CountDownLatch(1);
//        final String message = "Hazelcast Rocks!";
        MessageListener<String> messageListener = new MessageListener<String>() {
            public void onMessage(Message<String> msg) {
//                if (msg.startsWith(message)) {
                System.out.println("Received " + msg + " at " + this);
                latch.countDown();
                cp.countDown();
//                }
            }
        };
        final String message = "message_" + messageListener.hashCode() + "_";
        topic.addMessageListener(messageListener);
        topic.publish(message + "1");
        cp.await();
        topic.removeMessageListener(messageListener);
        topic.publish(message + "2");
        Thread.sleep(50);
        assertEquals(1, latch.getCount());
    }

项目：health-and-care-developer-network 文件：MessageListenerManagerTest.java

@Test
public void testRemoveMessageListener() throws Exception {
    MessageListenerManager manager = new MessageListenerManager();
    String name = "default";
    assertTrue(manager.noListenerRegistered(name));
    MessageListener listener = new MessageListener<Object>() {

        public void onMessage(Message<Object> message) {
        }
    };
    manager.registerListener(name, listener);
    assertFalse(manager.noListenerRegistered(name));
    manager.removeListener(name, listener);
    assertTrue(manager.noListenerRegistered(name));
    manager.removeListener(name, listener);
    assertTrue(manager.noListenerRegistered(name));
}

项目：scalable-websocket-chat-with-hazelcast 文件：NewJoineeNotificationEventListener.java

@Override
public void onMessage(Message<NewJoineeNotification> msg) {
    System.out.println("HZ NewJoineeNotification Topic Listener invoked");
    ChatServer.getSessions().stream()
            .filter((sn) -> !sn.getUserProperties().get("user").equals(msg.getMessageObject().getNewJoinee()))
            .forEach((sn) -> sn.getAsyncRemote().sendObject(msg.getMessageObject()));
}

项目：scalable-websocket-chat-with-hazelcast 文件：LogoutNotificationEventListener.java

@Override
public void onMessage(Message<LogOutNotification> event) {
    System.out.println("HZ Logout notification Topic Listener invoked");
    LogOutNotification msg = event.getMessageObject();
    ChatServer.getSessions().stream()
                .filter((sn) -> sn.isOpen())
                .forEach((session) -> session.getAsyncRemote().sendObject(msg));

}

项目：lannister 文件：Sessions.java

@Override
public void onMessage(Message<Notification> message) {
    Notification notified = message.getMessageObject();

    Session session = get(notified.clientId());
    if (session == null || !session.isConnected(true)) { return; }

    session.sendPublish(notified.topic(), notified.message());// [MQTT-3.3.1-8],[MQTT-3.3.1-9]
}

项目：lannister 文件：SingleTopic.java

@Override
public void publish(E message) {
    final Date now = new Date();

    GlobalEventExecutor.INSTANCE.execute(() -> {
        Message<E> msg = new Message<E>(name, message, now.getTime(), null);
        messageListeners.values().forEach(c -> c.onMessage(msg));
    });
}

项目：reactive-data 文件：ModelCombinerComponent.java

@Override
public void onMessage(Message<Byte> message) {
  log.debug("Message received from Member:: ["+message.getPublishingMember()+"] "+message.getMessageObject());
  switch (message.getMessageObject()) 
  {
    case DUMP_MODEL_REQ:
      dumpClassifierSnapshot();
      break;
    case DUMP_MODEL_RES:
      notifyIfProcessing();
      break;  
    default:
      break;
  }
}

项目：reactive-data 文件：AbstractFileSharingAgent.java

@Override
public void onMessage(Message<Byte> message) 
{
  if(message.getPublishingMember().localMember())
    return;

  switch(message.getMessageObject())
  {
    case SEND_FILE_ACK:
    if (awaitingFileSendAck) {
      latch.countDown();
    }
    break;
    case SEND_FILE:
      startFileConsumingTask();
      break;
    case RECV_FILE_ACK:
      if (awaitingFileReceiptAck) {
        latch.countDown();
      }
      break;
    case RECV_FILE_ERR:
      if (awaitingFileReceiptAck) {
        sharingErrorCount.incrementAndGet();
        fileReceiptErrored.add(message.getPublishingMember());
        latch.countDown();
        log.error("*** RECV_FILE_ERR ***");
      }

      break;
      default: break;
  }

}

项目：hazelcast-hibernate5 文件：LocalRegionCache.java

protected MessageListener<Object> createMessageListener() {
    return new MessageListener<Object>() {

        @Override
        public void onMessage(final Message<Object> message) {
                maybeInvalidate(message.getMessageObject());
        }
    };
}

项目：eet.osslite.cz 文件：HazelcastConfiguration.java

private void setTopicConfig(Config config) {
    TopicConfig topicConfig = config.getTopicConfig("yourTopicName");
    topicConfig.setGlobalOrderingEnabled(true);
    topicConfig.setStatisticsEnabled(true);

    MessageListener<String> implementation = new MessageListener<String>() {
        @Override
        public void onMessage(Message<String> message) {
            // process the message
        }
    };
    topicConfig.addMessageListenerConfig(new ListenerConfig(implementation));
}

项目：hazelcast-hibernate 文件：LocalRegionCache.java

protected MessageListener<Object> createMessageListener() {
    return new MessageListener<Object>() {
        public void onMessage(final Message<Object> message) {
            if (!message.getPublishingMember().localMember()) {
                maybeInvalidate(message.getMessageObject());
            }
        }
    };
}

项目：hazelcast-hibernate 文件：LocalRegionCache.java

protected MessageListener<Object> createMessageListener() {
    return new MessageListener<Object>() {
        public void onMessage(final Message<Object> message) {
            if (!message.getPublishingMember().localMember()) {
                maybeInvalidate(message.getMessageObject());
            }
        }
    };
}

项目：Camel 文件：HazelcastTopicConsumerTest.java

@Test
@SuppressWarnings("unchecked")
public void receive() throws InterruptedException {
    MockEndpoint out = getMockEndpoint("mock:received");
    out.expectedMessageCount(1);

    final Message<String> msg = new Message<String>("foo", "foo", new java.util.Date().getTime(), null);
    argument.getValue().onMessage(msg);

    assertMockEndpointsSatisfied(2000, TimeUnit.MILLISECONDS);

    this.checkHeaders(out.getExchanges().get(0).getIn().getHeaders(), HazelcastConstants.RECEIVED);
}

项目：DSC 文件：HazelcastService.java

@Override
public void progressRoleMessage(Message<RoleMessage> message) {//处理角色请求订阅消息
    RoleMessage roleMessage = message.getMessageObject();
    String switchId = roleMessage.SwitchId;
    DatapathId dpid = DatapathId.of(switchId);// 得到请求交换机机id

    IOFSwitch sw = switchService.getSwitch(dpid);// 得到交换机
    OFControllerRole controllerRole = parseRole(roleMessage.Role);
    sw.writeRequest(sw.getOFFactory()
            .buildRoleRequest()
            .setGenerationId(U64.ZERO)
            .setRole(controllerRole).
            build());

}

项目：dolphin-platform 文件：ExternalServer.java

private void addConsumer(HazelcastInstance instance, String topicName) {
    final ITopic<CustomEventFormat> topic = instance.getTopic(topicName);
    topic.addMessageListener(new MessageListener<CustomEventFormat>() {

        @Override
        public void onMessage(Message<CustomEventFormat> message) {
            System.out.println("Received: " + message.getMessageObject().getMyMessage());
        }
    });
}

项目：age3-nanorobots 文件：DefaultTopologyService.java

@Override public void onMessage(final Message<TopologyMessage> message) {
    log.debug("Distributed event: {}", message);
    final TopologyMessage topologyMessage = message.getMessageObject();
    switch (topologyMessage.getType()) {
        case TOPOLOGY_SELECTED:
            service.fire(Event.TOPOLOGY_CONFIGURED);
            break;
    }
}