/* Ċ Ġ* ĠCopyright Ġ 2 0 1 8 - 2 0 2 1 ĠSIP 3 . IO , ĠCorp . Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ Ċ package Ġio . sip 3 . sal to . ce . rtp r Ċ Ċ import Ġio . sip 3 . commons . domain . payload . R tp Report Payload Ċ import Ġio . sip 3 . commons . micro meter . Metrics Ċ import Ġio . sip 3 . commons . util . format Ċ import Ġio . sip 3 . commons . vertx . annotations . Instance Ċ import Ġio . sip 3 . commons . vertx . util . local Send Ċ import Ġio . sip 3 . sal to . ce . Attributes Ċ import Ġio . sip 3 . sal to . ce . Routes CE Ċ import Ġio . sip 3 . sal to . ce . attributes . Attributes Registry Ċ import Ġio . vertx . core . Abstract Vert icle Ċ import Ġio . vertx . core . json . JsonObject Ċ import Ġmu . Kotlin Logging Ċ import Ġjava . time . format . DateTime Formatter Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ /** Ċ Ġ* ĠHandles ĠR tp r Session Ċ Ġ*/ Ċ @ Instance Ċ open Ġclass ĠR tp r Session Handler Ġ: ĠAbstract Vert icle () Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠKotlin Logging . logger Ġ{} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĊĠĠĠĠĠĠĠ Ġconst Ġval ĠRE PORTS Ġ= Ġ"_ reports " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠBAD _ REPORT S Ġ= Ġ"_ bad - reports " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠBAD _ REPORT S _ FR ACTION Ġ= Ġ"_ bad - reports - fraction " ĊĊĠĠĠĠĠĠĠ Ġconst Ġval ĠD URATION Ġ= Ġ"_ duration " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġvar Ġtime Suffix : ĠDateTime Formatter Ġ= ĠDateTime Formatter . of Pattern (" yyyy MM dd ") ĊĊĠĠĠ Ġprivate Ġvar Ġmin Expected Packets Ġ= Ġ 1 0 0 ĊĊĠĠĠ Ġprivate Ġvar Ġrecord Ip Addresses Attributes Ġ= Ġfalse ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġattributes Registry : ĠAttributes Registry ĊĊĠĠĠ Ġoverride Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠ Ġconfig (). getString (" time - suffix ") ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtime Suffix Ġ= ĠDateTime Formatter . of Pattern ( it ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġconfig (). get JsonObject (" media ") ?. get JsonObject (" rtp - r ") ?. let Ġ{ Ġconfig Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconfig . getInteger (" min - expected - packets ") ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmin Expected Packets Ġ= Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġconfig (). get JsonObject (" attributes ") ?. getBoolean (" record - ip - addresses ") ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecord Ip Addresses Attributes Ġ= Ġit ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġattributes Registry Ġ= ĠAttributes Registry ( vertx , Ġconfig ()) ĊĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). local Consumer < R tp r Session >( Routes CE . rtp r Ġ+ Ġ"_ session ") Ġ{ Ġevent Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġr tp r Session Ġ= Ġevent . body () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhandle ( rtp r Session ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error ( e ) Ġ{ Ġ" R tp r Session Handler Ġ' handle ()' Ġfailed ." Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġopen Ġfun Ġhandle ( session : ĠR tp r Session ) Ġ{ ĊĠĠĠĠĠĠĠ Ġwrite Attributes ( session ) ĊĊĠĠĠĠĠĠĠ Ġval Ġprefix Ġ= Ġwhen Ġ( session . source ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠR tp Report Payload . SOURCE _ R TP Ġ-> Ġ" rtp r _ rtp " ĊĠĠĠĠĠĠĠĠĠĠĠ ĠR tp Report Payload . SOURCE _ RT CP Ġ-> Ġ" rtp r _ rt cp " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠIllegalArgumentException (" Unsupported ĠR tp r Session Ġsource : Ġ'${ session . source }' ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠwriteTo Database ("${ prefix }_ index ", Ġsession ) ĊĠĠĠĠĠĠĠ Ġcalculate Metrics ( prefix , Ġsession ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġopen Ġfun Ġwrite Attributes ( session : ĠR tp r Session ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġattributes Ġ= Ġmutable Map Of < String , ĠAny >(). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsrc Ġ= Ġsession . src Addr ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Attributes . src _ addr , Ġif Ġ( record Ip Addresses Attributes ) Ġsrc . addr Ġelse Ġ"") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsrc . host ?. let Ġ{ Ġput ( Attributes . src _ host , Ġit ) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdst Ġ= Ġsession . dst Addr ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Attributes . dst _ addr , Ġif Ġ( record Ip Addresses Attributes ) Ġdst . addr Ġelse Ġ"") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdst . host ?. let Ġ{ Ġput ( Attributes . dst _ host , Ġit ) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( session . forward Ġ?: Ġsession . reverse )?. let Ġ{ Ġstream Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream . mos ?. let Ġ{ Ġput ( Attributes . mos , Ġit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream . r Factor ?. let Ġ{ Ġput ( Attributes . r _ factor , Ġit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Attributes . bad _ report _ fraction , Ġsession . bad Report Fraction ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Attributes . one _ way , Ġsession . is One Way ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġprefix Ġ= Ġwhen Ġ( session . source ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠR tp Report Payload . SOURCE _ R TP Ġ-> Ġ" rtp " ĊĠĠĠĠĠĠĠĠĠĠĠ ĠR tp Report Payload . SOURCE _ RT CP Ġ-> Ġ" rt cp " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠIllegalArgumentException (" Unsupported ĠR tp r Session Ġsource : Ġ'${ session . source }' ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġattributes Registry . handle ( prefix , Ġattributes ) ĊĊĠĠĠĠĠĠĠ Ġsession . code cs . forEach Ġ{ Ġattributes Registry . handle ( prefix , ĠAttributes . codec , Ġit ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġopen Ġfun Ġcalculate Metrics ( prefix : ĠString , Ġsession : ĠR tp r Session ) Ġ{ ĊĠĠĠĠĠĠĠ Ġsession . apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġattributes Ġ= Ġmutable Map Of < String , ĠAny >(). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsrc Addr . host ?. let Ġ{ Ġput (" src _ host ", Ġit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdst Addr . host ?. let Ġ{ Ġput (" dst _ host ", Ġit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcodecs . first OrNull ()?. let Ġ{ Ġput (" codec ", Ġit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics . summary ( prefix Ġ+ ĠRE PORTS , Ġattributes ). record ( report Count . toDouble ()) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics . summary ( prefix Ġ+ ĠBAD _ REPORT S , Ġattributes ). record ( bad Report Count . toDouble ()) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics . summary ( prefix Ġ+ ĠBAD _ REPORT S _ FR ACTION , Ġattributes ). record ( bad Report Fraction ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠMetrics . timer ( prefix Ġ+ ĠD URATION , Ġattributes ). record ( duration , ĠTimeUnit . MILLISECONDS ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġopen Ġfun ĠwriteTo Database ( prefix : ĠString , Ġsession : ĠR tp r Session ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcollection Ġ= Ġprefix Ġ+ Ġ"_" Ġ+ Ġtime Suffix . format ( session . createdAt ) ĊĊĠĠĠĠĠĠĠ Ġval Ġoperation Ġ= ĠJsonObject (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput (" document ", ĠJsonObject (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" created _ at ", Ġsession . createdAt ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" terminated _ at ", Ġsession . terminated At ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsrc Ġ= Ġsession . src Addr ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" src _ addr ", Ġsrc . addr ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" src _ port ", Ġsrc . port ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsrc . host ?. let Ġ{ Ġput (" src _ host ", Ġit ) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdst Ġ= Ġsession . dst Addr ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" dst _ addr ", Ġdst . addr ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" dst _ port ", Ġdst . port ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdst . host ?. let Ġ{ Ġput (" dst _ host ", Ġit ) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" call _ id ", Ġsession . media Control . call Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" caller ", Ġsession . media Control . caller ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" callee ", Ġsession . media Control . callee ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" duration ", Ġsession . duration ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" report _ count ", Ġsession . report Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" bad _ report _ count ", Ġsession . bad Report Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" bad _ report _ fraction ", Ġsession . bad Report Fraction ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" one _ way ", Ġsession . is One Way ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdirections Ġ= Ġmutable ListOf < String >(). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( session . forward Ġ!= Ġnull ) Ġadd (" out ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( session . reverse Ġ!= Ġnull ) Ġadd (" in ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" direction ", Ġdirections ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreports Ġ= Ġmutable ListOf < R tp Report Payload >(). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsession . forward ?. report ?. let Ġ{ Ġadd ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsession . reverse ?. report ?. let Ġ{ Ġadd ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" mos ", Ġreports . map Ġ{ Ġit . mos . toDouble () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" r _ factor ", Ġreports . map Ġ{ Ġit . r Factor . toDouble () Ġ}) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" payload _ type ", Ġreports . map Ġ{ Ġit . payload Type . toInt () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" codec ", Ġreports . map Ġ{ Ġit . codec Name Ġ?: Ġ" UNDEFINED (${ it . payload Type })" Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" ss rc ", Ġreports . map Ġ{ Ġit . ss rc Ġ}) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" packets ", ĠJsonObject (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" expected ", Ġreports . map Ġ{ Ġit . expected Packet Count Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" received ", Ġreports . map Ġ{ Ġit . received Packet Count Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" lost ", Ġreports . map Ġ{ Ġit . lost Packet Count Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" rejected ", Ġreports . map Ġ{ Ġit . rejected Packet Count Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" j itter ", ĠJsonObject (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" last ", Ġreports . map Ġ{ Ġit . last J itter . toDouble () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" avg ", Ġreports . map Ġ{ Ġit . avg J itter . toDouble () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" min ", Ġreports . map Ġ{ Ġit . min J itter . toDouble () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" max ", Ġreports . map Ġ{ Ġit . max J itter . toDouble () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput (" fraction _ lost ", Ġreports . map Ġ{ Ġit . fraction Lost . toDouble () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġvert x . event Bus (). local Send ( Routes CE . mongo _ bulk _ writer , ĠPair ( collection , Ġoperation )) ĊĠĠĠ Ġ} Ċ }