package Ġmonitor Ċ Ċ import Ġjava . net . Inet SocketAddress Ċ import Ġjava . net . Server Socket Ċ import Ġjava . net . Socket Ċ Ċ private Ġfun Ġbuild Socket Map ( this NodeId : ĠNode Id , Ġcluster Config : ĠList < Inet SocketAddress >): ĠMap < NodeId , ĠSocket > Ġ{ ĊĠĠĠ Ġval Ġserver Socket Ġ= ĠServer Socket ( cluster Config [ this NodeId ]. port ) ĊĠĠĠ Ġprintln (" Bound Ġaddress : Ġ${ server Socket . local SocketAddress }") ĊĠĠĠ Ġval Ġlower Nodes Sockets Ġ= Ġconnect To Nodes ( cluster Config . sub List ( 0 , Ġthis NodeId )) ĊĠĠĠ Ġval Ġhigher Nodes Sockets Ġ= Ġaccept Nodes Connections ( this NodeId Ġ+ Ġ 1 Ġuntil Ġcluster Config . size , Ġserver Socket ) ĊĠĠĠ Ġreturn Ġ( lower Nodes Sockets Ġ+ Ġhigher Nodes Sockets ). toMap () Ċ } Ċ Ċ private Ġfun Ġconnect To Nodes ( addresses : ĠList < Inet SocketAddress >): ĠList < Pair < NodeId , ĠSocket >> Ġ= ĊĠĠĠĠĠĠĠ Ġaddresses . map Indexed Ġ{ ĠnodeId , Ġsocket Address Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Connecting Ġto Ġ$ socket Address ...") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠnodeId Ġto ĠSocket ( socket Address . address , Ġsocket Address . port ) ĊĠĠĠĠĠĠĠ Ġ} Ċ Ċ private Ġfun Ġaccept Nodes Connections ( nodeId Range : ĠInt Range , Ġserver Socket : ĠServer Socket ): ĠList < Pair < NodeId , ĠSocket >> Ġ= ĊĠĠĠĠĠĠĠ ĠnodeId Range . map Ġ{ ĠnodeId Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsocket Ġ= Ġserver Socket . accept () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln ("${ socket . remote SocketAddress } Ġconnected !") ĊĠĠĠĠĠĠĠĠĠĠĠ ĠnodeId Ġto Ġsocket ĊĠĠĠĠĠĠĠ Ġ} Ċ Ċ fun Ġtcp Cluster Socket ( this NodeId : ĠNode Id , Ġcluster Config : ĠList < Inet SocketAddress >): ĠCluster Socket Ġ{ ĊĠĠĠ Ġval Ġraw Socket Map Ġ= Ġbuild Socket Map ( this NodeId , Ġcluster Config ) ĊĊĠĠĠ Ġprintln (" Socket Ġmap :") ĊĠĠĠ Ġraw Socket Map . forEach Ġ{ ĠnodeId , Ġsocket Ġ-> ĊĠĠĠĠĠĠĠ Ġprintln ("$ nodeId : Ġ${ socket . local SocketAddress } Ġ-> Ġ${ socket . remote SocketAddress }") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġsocket Map Ġ= Ġraw Socket Map . map Values Ġ{ ĠMessage Socket ( it . value ) Ġ} ĊĊĠĠĠ Ġreturn ĠCluster Socket ( this NodeId , Ġsocket Map ) Ċ } Ċ