covid_tweets_processor.java

@Component
@Slf4j
@RequiredArgsConstructor
public class TweetsIngestHandler {

    /**
     * Text Analyzer Service
     */
    private final ITextAnalyzerService textAnalyzerService;

    /**
     *
     * @param newTweet
     * @param acknowledgment
     * @return
     */
    @StreamListener(AppStreamsConfig.TWEET_INGEST_CHANNEL)
    @SendTo(AppStreamsConfig.PROCESSED_TWEETS_CHANNEL)
    public TweetDTO onNewTweet(
            @Payload final TweetDTO newTweet,
            @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {

        final TextAnalysisResult analysisResult = textAnalyzerService.analyze(newTweet.getText());
        newTweet.setSentimentLabel(analysisResult.getSentiment().name());
        newTweet.setSentimentValue(analysisResult.getSentiment().getValue());
        newTweet.setEntityMentions(analysisResult.getEntityMentions());
        newTweet.setTokensAndNERTags(analysisResult.getTokensAndNERTags());

        if (acknowledgment != null) {
            acknowledgment.acknowledge();
        }

        log.debug("Sentiment For Current Tweet -> " + analysisResult.getSentiment().name());
        return newTweet;

    }
}