package Ġcom . github . im f log . gcp Ċ Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġcom . fasterxml . jackson . module . kotlin . readValue Ċ import Ġcom . google . api . g ax . core . No Credentials Provider Ċ import Ġcom . google . api . g ax . rpc . AlreadyExists Exception Ċ import Ġcom . google . api . g ax . rpc . Fixed Transport Channel Provider Ċ import Ġcom . google . cloud . pubsub . v 1 . Ack Reply Consumer Ċ import Ġcom . google . cloud . pubsub . v 1 . Message Receiver Ċ import Ġcom . google . cloud . pubsub . v 1 . Subscriber Ċ import Ġcom . google . cloud . pubsub . v 1 . Subscription Admin Client Ċ import Ġcom . google . pubsub . v 1 . Project Subscription Name Ċ import Ġcom . google . pubsub . v 1 . Project Topic Name Ċ import Ġcom . google . pubsub . v 1 . Pub sub Message Ċ import Ġcom . google . pubsub . v 1 . Push Config Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . stereotype . Component Ċ import Ġjavax . annotation . Post Construct Ċ import Ġkotlin . concurrent . thread ĊĊ Ċ @ Component Ċ class ĠPokemon Consumer ( ĊĠĠĠ Ġprivate Ġval Ġpokemon Repository : ĠPokemon Repository , ĊĠĠĠ Ġprivate Ġval Ġchannel Provider : ĠFixed Transport Channel Provider , ĊĠĠĠ Ġprivate Ġval Ġsubscription Admin Client : ĠSubscription Admin Client , ĊĠĠĠ Ġprivate Ġval ĠobjectMapper : ĠObjectMapper Ċ ) Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval ĠLOGGER Ġ= ĠLoggerFactory . getLogger ( Pokemon Consumer :: class . java ) ĊĊĠĠĠĠĠĠĠ Ġ// ĠTODO Ġ: ĠMove Ġthis Ġin Ġconfiguration Ġ? ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval ĠSUB SCRIPTION _ NAME Ġ= Ġ" poke - subscription " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Post Construct ĊĠĠĠ Ġ// ĠTODO Ġ: ĠThis Ġmethod Ġis Ġa Ġbit Ġtoo Ġlinked Ġto Ġconfiguration ĊĠĠĠ Ġ// ĠTODO : ĠTry Ġto Ġextract Ġthe ĠSubscriber Ġ... ĊĠĠĠ Ġfun Ġsave Ca ught Pokemon Info () Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠDo Ġwe Ġneed Ġthis Ġhere Ġ? ĊĠĠĠĠĠĠĠ Ġcreate Subscription () ĊĠĠĠĠĠĠĠ Ġ// Ġ---------------- ------ ĊĊĠĠĠĠĠĠĠ ĠLOGGER . info (" Starting Ġconsumption Ġof Ġcaught Ġpokemon ") ĊĠĠĠĠĠĠĠ Ġval Ġsubscription Name Ġ= ĠProject Subscription Name . of ( PROJECT _ ID , ĠSUB SCRIPTION _ NAME ) ĊĠĠĠĠĠĠĠ Ġthread Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġsubscriber : ĠSubscriber ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO : ĠShould Ġwe Ġbuild Ġthe Ġsubscriber Ġin Ġthe Ġconfiguration Ġ? ĠExcept Ġfor Ġthe Ġfunction ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber Ġ= ĠSubscriber . newBuilder ( subscription Name , Ġprocess Message ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO Ġ: ĠPUT ĠTHIS ĠIN ĠCONFIG ĠFILE S ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Channel Provider ( channel Provider ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠTODO Ġ: ĠThis Ġalso Ġshould Ġcome Ġfrom Ġthe Ġconfiguration ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set Credentials Provider ( No Credentials Provider . create ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . start Async (). await Running () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAllow Ġthe Ġsubscriber Ġto Ġrun Ġin def initely Ġunless Ġan Ġun recover able Ġerror Ġoccurs ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber . await Terminated () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠStop Ġreceiving Ġmessages ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscriber ?. stop Async () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġprocess Message (): ĠMessage Receiver Ġ= ĠMessage Receiver Ġ{ Ġmessage , Ġ_: ĠAck Reply Consumer Ġ-> ĊĠĠĠĠĠĠĠ Ġval Ġpokemon Ġ= ĠobjectMapper . readValue < Pokemon >( message . data . toByteArray ()) ĊĠĠĠĠĠĠĠ Ġpokemon Repository . insert Pokemon ( pokemon ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Subscription () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= ĠProject Topic Name . of ( PROJECT _ ID , ĠTOP IC _ NAME ) ĊĠĠĠĠĠĠĠ Ġval Ġsubscription Ġ= ĠProject Subscription Name . of ( PROJECT _ ID , ĠSUB SCRIPTION _ NAME ) ĊĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription Admin Client . create Subscription ( subscription , Ġtopic , ĠPush Config . getDefaultInstance (), Ġ 1 0 0 ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠAlready Exists Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġthis Ġis Ġfine , Ġalready Ġcreated ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ