/* Ċ Ġ* ĠCopyright Ġ 2 0 1 8 - 2 0 2 2 Ġ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 . capt ain . ce . pipeline Ċ Ċ import Ġio . sip 3 . capt ain . ce . Routes CE Ċ import Ġio . sip 3 . capt ain . ce . domain . Packet Ċ import Ġio . sip 3 . capt ain . ce . util . S mpp Util Ċ import Ġio . sip 3 . commons . Packet Types Ċ import Ġio . sip 3 . commons . domain . payload . ByteArray Payload Ċ import Ġio . sip 3 . commons . domain . payload . En codable Ċ import Ġio . sip 3 . commons . util . getBytes Ċ import Ġio . sip 3 . commons . vertx . util . local Send Ċ import Ġio . vertx . core . Vert x Ċ import Ġio . vertx . core . json . JsonObject Ċ Ċ /** Ċ Ġ* ĠHandles ĠS MP P Ġpackets Ċ Ġ*/ Ċ class ĠS mpp Handler ( vertx : ĠVert x , Ġconfig : ĠJsonObject , Ġbulk Operations Enabled : ĠBoolean ) Ġ: ĠHandler ( vertx , Ġconfig , Ġbulk Operations Enabled ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġpackets Ġ= Ġmutable ListOf < Packet >() ĊĠĠĠ Ġprivate Ġvar Ġbulk Size Ġ= Ġ 1 ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( bulk Operations Enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconfig . get JsonObject (" s mpp ") ?. let Ġ{ Ġs mpp Config Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġs mpp Config . getInteger (" bulk - size ") ?. let Ġ{ Ġbulk Size Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġon Packet ( packet : ĠPacket ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġbuffer Ġ= Ġ( packet . payload Ġas ĠEn codable ). encode () ĊĊĠĠĠĠĠĠĠ Ġif Ġ( S mpp Util . check Min P du Length ( buffer )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoffset Ġ= Ġbuffer . reader Index () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlength Ġ= Ġbuffer . get Unsigned Int ( offset ). toInt () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( buffer . readable Bytes () Ġ>= Ġlength ) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= Ġbuffer . get Unsigned Int ( offset Ġ+ Ġ 4 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( S mpp Util . is P du Command ( command )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġp Ġ= Ġpacket . clone () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġp . apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprotocol Code Ġ= ĠPacket Types . S MP P ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpayload Ġ= Ġrun Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbytes Ġ= Ġbuffer . getBytes ( offset , Ġlength ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ run ĠByteArray Payload ( bytes ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpackets . add ( p ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( packets . size Ġ>= Ġbulk Size ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). local Send ( Routes CE . encoder , Ġpackets . toList ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpackets . clear () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbuffer . reader Index ( offset Ġ+ Ġlength ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( buffer . readable Bytes () Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġon Packet ( packet ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }