EmojiCounterBolt extends ChatAlyticsBaseBolt { @Override public void execute(Tuple input, BasicOutputCollector collector) { LOG.debug("Got tuple: {}", input); FatMessage fatMessage = (FatMessage) input.getValue(0); List<EmojiEntity> emojis = getEmojisFromMessage(fatMessage); for (EmojiEntity emoji : emojis) { emojiDao.persistEmoji(emoji); collector.emit(new Values(emoji)); } }  @Override void prepare(ChatAlyticsConfig config, @SuppressWarnings("rawtypes") Map conf,
                        TopologyContext context); @Override void execute(Tuple input, BasicOutputCollector collector); @Override void declareOutputFields(OutputFieldsDeclarer fields); @Override void cleanup(); static final String BOLT_ID; }
@Test public void testExecute() { Message message = new Message(mentionTime, "randomFrom", "randomUserId", String.format("test message with :%s:", emoji), "randomRoomId", MessageType.MESSAGE); FatMessage fatMessage = new FatMessage(message, user, room); List<Object> values = Lists.newArrayList(fatMessage); TopologyContext context = mock(TopologyContext.class); Fields fields = mock(Fields.class); when(fields.size()).thenReturn(1); when(context.getComponentOutputFields(anyString(), anyString())).thenReturn(fields); Tuple input = new TupleImpl(context, values, 0, "stream-id"); BasicOutputCollector collector = mock(BasicOutputCollector.class); ChatAlyticsConfig config = new ChatAlyticsConfig(); config.computeConfig.apiRetries = 0; config.persistenceUnitName = "chatalytics-db-test"; Map<Object, Object> stormConf = Maps.newHashMapWithExpectedSize(1); stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config)); underTest.prepare(stormConf, context); underTest.execute(input, collector); verify(collector).emit(any(Values.class)); }