EntityExtractionBolt extends ChatAlyticsBaseBolt { @Override public void execute(Tuple input, BasicOutputCollector collector) { LOG.debug("Got tuple: {}", input); FatMessage fatMessage = (FatMessage) input.getValue(0); List<ChatEntity> entities = extractEntities(fatMessage); for (ChatEntity entity : entities) { entityDao.persistEntity(entity); collector.emit(new Values(entity)); } }  @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; static final String CHAT_ENTITY_FIELD_STR; }
@Test public void testExecute() { String ent1 = "Jane Doe"; String ent2 = "Mount Everest"; Message msg = new Message(DateTime.now(), "jane", "u1", String.format("Today, %s is going to climb %s", ent1, ent2), "r1", MessageType.MESSAGE); FatMessage fatMessage = new FatMessage(msg, user, room); List<Object> values = Lists.newArrayList(fatMessage); Tuple input = new TupleImpl(context, values, 0, "stream-id"); underTest.execute(input, collector); verify(collector, times(2)).emit(any(Values.class)); }