SlackBackfillSpout extends BaseRichSpout { @VisibleForTesting protected void backfillRooms(Map<String, User> users, Map<String, Room> rooms, Interval runInterval) { Set<String> processedRoomNames = Sets.newHashSet(rooms.keySet()); int skippedUnknownMessages = 0; LOG.info("Backfilling {} rooms", rooms.size()); int roomNum = 0; for (Room room : rooms.values()) { roomNum++; processedRoomNames.add(room.getName()); List<Message> messages = slackDao.getMessages(runInterval.getStart(), runInterval.getEnd(), room); for (Message message : messages) { User user = users.get(message.getFromUserId()); if (message.getType() == MessageType.UNKNOWN) { LOG.debug("Skipping unkown message type. {}", message); skippedUnknownMessages++; continue; } else if (user == null && message.getType() == MessageType.BOT_MESSAGE) { user = new User(message.getFromUserId(), null, false, false, true, message.getFromName(), message.getFromName(), null, DateTime.now(), null, null, null, null, null); } if (user == null) { LOG.warn("Can't find user with userId: {}. Skipping", message.getFromUserId()); continue; } FatMessage fatMessage = new FatMessage(message, user, room); collector.emit(new Values(fatMessage)); } logProgress(roomNum, rooms.size()); } LOG.info("Finished backfilling. Skipped {} unknown msgs. Processed {} rooms. They were: {}", skippedUnknownMessages, processedRoomNames.size(), processedRoomNames); }  @Override void open(@SuppressWarnings("rawtypes") Map conf, TopologyContext context,
                     SpoutOutputCollector collector); @Override void nextTuple(); @Override void declareOutputFields(OutputFieldsDeclarer fields); @Override void close(); static final String SPOUT_ID; static final String BACKFILL_SLACK_MESSAGE_FIELD_STR; }
@Test public void testBackfillRooms() { stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config)); IChatAlyticsDAO dbDao = mock(IChatAlyticsDAO.class); IChatApiDAO slackDao = mock(IChatApiDAO.class); underTest.open(chatConfig, slackDao, dbDao, context, collector); Map<String, User> users = ImmutableMap.of("u1", new User("u1", "email", false, false, false, "name", "mention_name", null, DateTime.now(), DateTime.now(), null, null, null, null)); Room room = new Room("r1", "room", null, DateTime.now(), DateTime.now(), null, false, false, null, null); Map<String, Room> rooms = ImmutableMap.of("r1", room); Interval interval = new Interval(DateTime.now().minusDays(1), DateTime.now()); Message message = new Message(DateTime.now(), "from", "u1", "test message", "r1", MessageType.MESSAGE); List<Message> messages = Lists.newArrayList(message); when(slackDao.getMessages(interval.getStart(), interval.getEnd(), room)).thenReturn(messages); underTest.backfillRooms(users, rooms, interval); verify(slackDao).getMessages(interval.getStart(), interval.getEnd(), room); verifyNoMoreInteractions(slackDao); verify(collector).emit(any(Values.class)); verifyNoMoreInteractions(collector); }