package Ġfr anz Ċ Ċ import Ġfr anz . engine . Consumer Actor Factory Ċ import Ġfr anz . engine . Worker Function Ċ import Ġfr anz . engine . kafka _ one . Kafka Consumer Actor Factory Ċ import Ġkotlinx . coroutines . Coroutine Dispatcher Ċ import Ġkotlinx . coroutines . Coroutine Name Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . as Coroutine Dispatcher Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . Array BlockingQueue Ċ import Ġjava . util . concurrent . ThreadPoolExecutor Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ private Ġval Ġstring Des er Ġ= Ġ" org . apache . kafka . common . serialization . String Deserializer " Ċ private Ġval ĠbyteArray Des er Ġ= Ġ" org . apache . kafka . common . serialization . ByteArray Deserializer " Ċ private Ġval Ġvalue Des er Key Ġ= Ġ" value . de serializer " Ċ Ċ private Ġtype alias ĠRunning Function < T , ĠU > Ġ= Ġsuspend ĠJob DSL < T , ĠU >. () Ġ-> ĠJob Status Ċ private Ġtype alias ĠP ip ed Worker Function < T , ĠU > Ġ= Ġsuspend Ġ( Job State < Message < T , ĠU >>) Ġ-> ĠJob Status Ċ Ċ private Ġsuspend Ġfun Ġ< T , ĠU > Ġrunning Worker ( fn : ĠRunning Function < T , ĠU >): ĠWorker Function < T , ĠU > Ġ= Ġ{ ĊĠĠĠ Ġfn ( Job DSL ( it )) Ċ } Ċ Ċ private Ġsuspend Ġfun Ġ< T , ĠU > Ġpip ed Worker ( fn : ĠP ip ed Worker Function < T , ĠU >, Ġintercept ors : ĠList < Worker Interceptor >): ĠWorker Function < T , ĠU > Ġ= Ġ{ ĊĠĠĠ Ġfn ( Job State ( it , ĠStack (), ĠStack (), intercept ors . toList ())) Ċ } Ċ Ċ /** Ċ Ġ* ĠThe Ġnumber Ġof Ġthreads Ġto Ġkeep Ġin Ġthe Ġpool , Ġeven Ġif Ġthey Ġare Ġidle Ċ Ġ*/ Ċ private Ġconst Ġval ĠTHREAD _ POOL _ MIN _ SIZE : ĠInt Ġ= Ġ 1 Ċ /** Ċ Ġ* ĠThe Ġmaximum Ġnumber Ġof Ġthreads Ġto Ġallow Ġin Ġthe Ġpool Ċ Ġ*/ Ċ private Ġconst Ġval ĠTHREAD _ POOL _ MAX _ SIZE : ĠInt Ġ= Ġ 2 Ċ /** Ċ Ġ* ĠWhen Ġthe Ġnumber Ġof Ġthreads Ġis Ġgreater Ġthan Ġthe Ġ[ THREAD _ POOL _ MAX _ SIZE ], Ċ Ġ* Ġthis Ġis Ġthe Ġmaximum Ġtime Ġin Ġseconds Ġthat Ġexcess Ġidle Ġthreads Ġwill Ġwait Ċ Ġ* Ġfor Ġnew Ġtasks Ġbefore Ġterminating Ċ Ġ*/ Ċ private Ġconst Ġval ĠDEFAULT _ THREAD _ POOL _ KEEP _ AL IVE _ TIME _ SECONDS : ĠLong Ġ= Ġ 3 0 Ċ Ċ private Ġfun Ġcreate Default Scope (): ĠCor outine Scope Ġ= ĊĠĠĠ Ġcreate Thread Scope ( THREAD _ POOL _ MIN _ SIZE , ĠTHREAD _ POOL _ MAX _ SIZE , ĠDEFAULT _ THREAD _ POOL _ KEEP _ AL IVE _ TIME _ SECONDS ) Ċ Ċ fun Ġcreate Thread Scope ( ĊĠĠĠ Ġmin PoolSize : ĠInt Ġ= ĠTHREAD _ POOL _ MIN _ SIZE , ĊĠĠĠ Ġmax PoolSize : ĠInt Ġ= ĠTHREAD _ POOL _ MAX _ SIZE , ĊĠĠĠ Ġthread Pool KeepAlive : ĠLong Ġ= ĠDEFAULT _ THREAD _ POOL _ KEEP _ AL IVE _ TIME _ SECONDS Ċ ): ĠCor outine Scope Ġ{ ĊĠĠĠ Ġval Ġdispatcher : ĠCor outine Dispatcher Ġ= ĠThreadPool Executor ( ĊĠĠĠĠĠĠĠ Ġmin PoolSize , ĊĠĠĠĠĠĠĠ Ġmax PoolSize , ĊĠĠĠĠĠĠĠ Ġthread Pool KeepAlive , ĊĠĠĠĠĠĠĠ ĠTimeUnit . SECONDS , ĊĠĠĠĠĠĠĠ ĠArray BlockingQueue ( 5 0 ) ĊĠĠĠ Ġ). as Coroutine Dispatcher () ĊĠĠĠ Ġreturn ĠCor outine Scope ( dispatcher Ġ+ ĠCor outine Name (" fr anz - coroutine - scope ")) Ċ } Ċ Ċ data Ġclass ĠWorker Builder < T > Ġprivate Ġconstructor ( ĊĠĠĠ Ġprivate Ġval Ġfn : ĠWorker Function < String , ĠT >? Ġ= Ġnull , ĊĠĠĠ Ġprivate Ġval Ġopts : ĠMap < String , ĠAny > Ġ= Ġempty Map (), ĊĠĠĠ Ġprivate Ġval Ġtopics : ĠList < String > Ġ= Ġempty List (), ĊĠĠĠ Ġprivate Ġval Ġengine : ĠConsumer Actor Factory Ġ= ĠKafka Consumer Actor Factory , ĊĠĠĠ Ġprivate Ġval Ġintercept ors : ĠList < Worker Interceptor > Ġ= Ġempty List (), ĊĠĠĠ Ġprivate Ġval Ġcoroutine Scope : ĠCor outine Scope Ġ= Ġcreate Default Scope () Ċ ){ ĊĠĠĠ Ġsuspend Ġfun Ġhandler ( f : ĠWorker Function < String , ĠT >) Ġ= Ġcopy ( fn Ġ= Ġf ) ĊĠĠĠ Ġ@ Deprecated (" Use Ġpip ed Ġor Ġhandler Ġinstead ") ĊĠĠĠ Ġsuspend Ġfun Ġrunning ( fn : ĠRunning Function < String , ĠT >) Ġ= Ġhandler ( running Worker ( fn )) ĊĠĠĠ Ġsuspend Ġfun Ġhandle Pip ed ( fn : ĠP ip ed Worker Function < String , ĠT >) Ġ= Ġhandler ( pip ed Worker ( fn , Ġintercept ors )) ĊĠĠĠ Ġfun Ġsubscribed To ( var arg Ġnew Topics : ĠString ) Ġ= Ġcopy ( topics Ġ= Ġtopics Ġ+ Ġnew Topics ) ĊĠĠĠ Ġfun Ġsubscribed To ( topics : ĠCollection < String >): ĠWorker Builder < T > Ġ= Ġmerge ( this , Ġtopics . to Typed Array ()) ĊĠĠĠ Ġfun ĠgroupId ( id : ĠString ) Ġ= Ġoption (" group . id ", Ġid ) ĊĠĠĠ Ġfun Ġoption ( k : ĠString , Ġv : ĠAny ) Ġ= Ġoptions ( map Of ( k Ġto Ġv )) ĊĠĠĠ Ġfun Ġoptions ( new Opts : ĠMap < String , ĠAny >) Ġ= Ġcopy ( opts Ġ= Ġopts Ġ+ Ġnew Opts ) ĊĠĠĠ Ġfun Ġcoroutine Scope ( coroutine Scope : ĠCor outine Scope ): ĠWorker Builder < T > Ġ= Ġcopy ( coroutine Scope Ġ= Ġcoroutine Scope ) ĊĠĠĠ Ġfun Ġset Engine ( e : ĠConsumer Actor Factory ): ĠWorker Builder < T > Ġ= Ġcopy ( engine Ġ= Ġe ) ĊĊĠĠĠ Ġfun Ġinstall ( i : ĠWorker Interceptor ): ĠWorker Builder < T > Ġ= Ġcopy ( intercept ors Ġ= Ġintercept ors . to Mutable List (). also { Ġit . add ( i )}) ĊĊĠĠĠ Ġfun Ġget Intercept ors () Ġ= Ġintercept ors ĊĊĠĠĠ Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġc Ġ= Ġengine . create < String , ĠT >( opts , Ġtopics ) ĊĠĠĠĠĠĠĠ Ġc . create Worker ( fn !! , Ġcoroutine Scope ) ĊĠĠĠĠĠĠĠ Ġc . start () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġtail rec Ġfun Ġmerge ( builder : ĠWorker Builder < T >, Ġtopics : ĠArray < String >, Ġi : ĠInt Ġ= Ġ 0 ): ĠWorker Builder < T > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen ( i Ġ> Ġtopics . last Index ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtrue Ġ-> Ġbuilder ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfalse Ġ-> Ġmerge ( builder . subscribed To ( topics [ i ]), Ġtopics , Ġi + 1 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġof ByteArray Ġ= ĠWorker Builder < ByteArray >(). option ( value Des er Key , ĠbyteArray Des er ) ĊĠĠĠĠĠĠĠ Ġval Ġof String Ġ= ĠWorker Builder < String >(). option ( value Des er Key , Ġstring Des er ) ĊĠĠĠ Ġ} Ċ }