package Ġeu . wo jc ie ch zure k . example . messaging Ċ Ċ import Ġeu . wo jc ie ch zure k . example . service . Sink Service Ċ import Ġeu . wo jc ie ch zure k . example . Word Statistic Ċ import Ġeu . wo jc ie ch zure k . example . logger For Ċ import Ġeu . wo jc ie ch zure k . example . messaging . Word Statistic Sink . Sink . Companion . INPUT Ċ import Ġorg . springframework . cloud . stream . annotation . Enable Binding Ċ import Ġorg . springframework . cloud . stream . annotation . Input Ċ import Ġorg . springframework . cloud . stream . annotation . Stream Listener Ċ import Ġorg . springframework . messaging . Sub scri b able Channel Ċ import Ġorg . springframework . stereotype . Component Ċ import Ġreactor . core . publisher . Flux Ċ Ċ @ Component Ċ @ Enable Binding ( Word Statistic Sink . Sink :: class ) Ċ class ĠWord Statistic Sink ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġsink Service : ĠS ink Service < Word Statistic > Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= Ġlogger For ( java Class ) ĊĊĠĠĠ Ġ@ Stream Listener ( INPUT ) ĊĠĠĠ Ġfun Ġconsume ( message : ĠFlux < Word Statistic >) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġmessage . subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Incoming Ġmessage : Ġ{}", Ġit ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsink Service . send ( it ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Component ĊĠĠĠ Ġinterface ĠS ink Ġ{ ĊĊĠĠĠĠĠĠĠ Ġ@ Input ( INPUT ) ĊĠĠĠĠĠĠĠ Ġfun Ġinput (): ĠSub scri b able Channel ĊĊĠĠĠĠĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconst Ġval ĠINPUT Ġ= Ġ" word - statistic - topic - in " ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ