package Ġrun . q on tract . core . pattern Ċ Ċ import Ġrun . q on tract . core . Resolver Ċ import Ġrun . q on tract . core . Result Ċ import Ġrun . q on tract . core . bread Cr umb Ċ import Ġrun . q on tract . core . value . EmptyString Ċ import Ġrun . q on tract . core . value . Kafka Message Ċ import Ġrun . q on tract . core . value . StringValue Ċ Ċ const Ġval ĠK AF KA _ MESSAGE _ B READ CR UMB Ġ= Ġ" K AF KA - MESSAGE " Ċ Ċ data Ġclass ĠKafka Message Pattern ( val Ġtopic : ĠString Ġ= Ġ"", Ġval Ġkey : ĠPattern Ġ= ĠEmpty String Pattern , Ġval Ġvalue : ĠPattern Ġ= ĠString Pattern ) Ġ{ ĊĠĠĠ Ġfun Ġmatches ( message : ĠKafka Message , Ġresolver : ĠResolver ): ĠResult Ġ= Ġattempt ( K AF KA _ MESSAGE _ B READ CR UMB ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmatch Topics ( message ). if Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmatch Key ( message , Ġresolver ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. if Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmatch Value ( message , Ġresolver ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠContract Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġe . failure () ĊĠĠĠĠĠĠĠ Ġ}. bread Cr umb ( K AF KA _ MESSAGE _ B READ CR UMB ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġmatch Value ( message : ĠKafka Message , Ġresolver : ĠResolver ): ĠResult Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġparsed Value Ġ= Ġwhen Ġ( message . value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠString Value Ġ-> Ġvalue . parse ( message . value . string , Ġresolver ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġmessage . value ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġvalue . matches ( parsed Value , Ġresolver ) ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠContract Exception ) Ġ{ ĊĠĠĠĠĠĠĠ Ġe . failure () ĊĠĠĠ Ġ}. bread Cr umb (" VALUE ") ĊĊĠĠĠ Ġprivate Ġfun Ġmatch Key ( message : ĠKafka Message , Ġresolver : ĠResolver ): ĠResult Ġ= Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġparsed Key Ġ= Ġwhen Ġ( message . key ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠString Value Ġ-> Ġkey . parse ( message . key . string , Ġresolver ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġmessage . key ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġkey . matches ( parsed Key Ġ?: ĠEmpty String , Ġresolver ) ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠContract Exception ) Ġ{ ĊĠĠĠĠĠĠĠ Ġe . failure () ĊĠĠĠ Ġ}. bread Cr umb (" KEY ") ĊĊĠĠĠ Ġprivate Ġfun Ġmatch Topics ( message : ĠKafka Message ): ĠResult Ġ= Ġwhen Ġ( message . topic ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtopic Ġ-> ĠResult . Success () ĊĠĠĠĠĠĠĠ Ġelse Ġ-> ĠResult . Failure (" Expected Ġtopic Ġ$ topic , Ġgot Ġ$ message . topic "). bread Cr umb (" TOPIC ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġen comp asses ( other : ĠKafka Message Pattern , Ġthis Resolver : ĠResolver , Ġother Resolver : ĠResolver ): ĠResult Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġattempt ( K AF KA _ MESSAGE _ B READ CR UMB ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopics Should Match ( other ). if Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġkey . en comp asses ( other . key , Ġother Resolver , Ġthis Resolver ). bread Cr umb (" KEY ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. if Success Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalue . en comp asses ( other . value , Ġthis Resolver , Ġother Resolver ). bread Cr umb (" VALUE ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. bread Cr umb ( K AF KA _ MESSAGE _ B READ CR UMB ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġtopics Should Match ( other : ĠKafka Message Pattern ): ĠResult Ġ= Ġwhen Ġ( topic ) Ġ{ ĊĠĠĠĠĠĠĠ Ġother . topic Ġ-> ĠResult . Success () ĊĠĠĠĠĠĠĠ Ġelse Ġ-> ĠResult . Failure (" Expected Ġtopic Ġ$ topic , Ġgot Ġ${ other . topic } "). bread Cr umb (" TOPIC ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġnew Based On ( row : ĠRow , Ġresolver : ĠResolver ): ĠList < Kafka Message Pattern > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġnew Keys Ġ= Ġkey . new Based On ( row , Ġresolver ) ĊĠĠĠĠĠĠĠ Ġval Ġnew Values Ġ= Ġvalue . new Based On ( row , Ġresolver ) ĊĊĠĠĠĠĠĠĠ Ġreturn Ġnew Keys . flatMap Ġ{ Ġnew Key Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnew Values . map Ġ{ ĠnewValue Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠKafka Message Pattern ( topic , Ġnew Key , ĠnewValue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġgenerate ( resolver : ĠResolver ): ĠKafka Message Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ ĠKafka Message ( topic , Ġkey . generate ( resolver ) Ġas ĠString Value , Ġvalue . generate ( resolver ) Ġas ĠString Value ) Ċ } Ċ