SlackMessageSpout extends BaseRichSpout { @OnMessage public void onMessageEvent(Message message, Session session) { LOG.debug("Got event {}", message); if (filterMessage(message)) { LOG.debug("Filtering message dated {}", message.getDate()); return; } Map<String, User> users = slackDao.getUsers(); Map<String, Room> rooms = slackDao.getRooms(); User fromUser = users.get(message.getFromUserId()); if (fromUser == null && message.getType() == MessageType.BOT_MESSAGE) { fromUser = new User(message.getFromUserId(), null, false, false, true, message.getFromName(), message.getFromName(), null, DateTime.now(), null, null, null, null, null); } if (fromUser == null) { LOG.warn("Can't find user with userId: {}. Skipping", message.getFromUserId()); return; } Room room = rooms.get(message.getRoomId()); if (room == null && message.getRoomId() != null) { room = new Room(message.getRoomId(), message.getRoomId(), null, DateTime.now(DateTimeZone.UTC), null, null, false, true, null, null); } FatMessage fatMessage = new FatMessage(message, fromUser, room); unemittedMessages.add(fatMessage); } SlackMessageSpout(); @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context,
                     SpoutOutputCollector collector); @OnMessage void onMessageEvent(Message message, Session session); @OnError void onError(Throwable t); @Override void nextTuple(); @Override void declareOutputFields(OutputFieldsDeclarer fields); @Override void close(); static final String SPOUT_ID; static final String SLACK_MESSAGE_FIELD_STR; }
@Test public void testOnMessageEvent() throws Exception { WebSocketContainer mockSocketContainer = mock(WebSocketContainer.class); when(mockSocketContainer.connectToServer(underTest, WEB_SOCKET_TEST_URI)) .thenReturn(mock(Session.class)); JsonSlackDAO slackDao = mock(JsonSlackDAO.class); when(slackDao.getRealtimeWebSocketURI()).thenReturn(WEB_SOCKET_TEST_URI); underTest.open(chatConfig, slackDao, mockSocketContainer, mockContext, mockCollector); String userId = "U03AFSSD"; Map<String, User> users = ImmutableMap.of(userId, new User(userId, null, false, false, false, "name", "mention_name", null, DateTime.now(), DateTime.now(), null, null, null, null)); when(slackDao.getUsers()).thenReturn(users); Message triggerMessage = new Message(DateTime.now(), "Test User", userId, "test msg", "C09ADF43", MESSAGE); underTest.onMessageEvent(triggerMessage, mock(Session.class)); verify(slackDao).getUsers(); verify(slackDao).getRooms(); verify(slackDao).getRealtimeWebSocketURI(); verifyNoMoreInteractions(slackDao); underTest.nextTuple(); verify(mockCollector).emit(any(Values.class)); verifyNoMoreInteractions(mockCollector); underTest.nextTuple(); verifyNoMoreInteractions(mockCollector); }