@ Override @ SuppressWarnings ( "checkstyle:diamondoperatorforvariabledefinition" ) public synchronized void onNext ( final RemoteEvent < byte [ ] > value ) { LOG . log ( Level . FINER , "RemoteManager: {0} value: {1}" , new Object [ ] { this . name , value } ) ; final T decodedEvent = this . codec . decode ( value . getEvent ( ) ) ; final Class < ? extends T > clazz = ( Class < ? extends T > ) decodedEvent . getClass ( ) ; LOG . log ( Level . FINEST , "RemoteManager: {0} decoded event {1} :: {2}" , new Object [ ] { this . name , clazz . getCanonicalName ( ) , decodedEvent } ) ; // check remote identifier and message type final SocketRemoteIdentifier id = new SocketRemoteIdentifier ( ( InetSocketAddress ) value . remoteAddress ( ) ) ; final Tuple2 < RemoteIdentifier , Class < ? extends T > > tuple = new Tuple2 < RemoteIdentifier , Class < ? extends T > > ( id , clazz ) ; final EventHandler < ? super T > tupleHandler = this . tupleToHandlerMap . get ( tuple ) ; if ( tupleHandler != null ) { LOG . log ( Level . FINER , "Tuple handler: {0},{1}" , new Object [ ] { tuple . getT1 ( ) , tuple . getT2 ( ) . getCanonicalName ( ) } ) ; tupleHandler . onNext ( decodedEvent ) ; } else { final EventHandler < RemoteMessage < ? extends T > > messageHandler = this . msgTypeToHandlerMap . get ( clazz ) ; if ( messageHandler == null ) { final RuntimeException ex = new RemoteRuntimeException ( "Unknown message type in dispatch: " + clazz . getCanonicalName ( ) + " from " + id ) ; LOG . log ( Level . WARNING , "Unknown message type in dispatch." , ex ) ; throw ex ; } LOG . log ( Level . FINER , "Message handler: {0}" , clazz . getCanonicalName ( ) ) ; messageHandler . onNext ( new DefaultRemoteMessage <> ( id , decodedEvent ) ) ; } }
Dispatch message received from the remote to proper event handler .