@ Override public void receive ( VehicleMessage message ) { if ( message == null ) { return ; } if ( message instanceof KeyedMessage ) { KeyedMessage keyedMessage = message . asKeyedMessage ( ) ; mKeyedMessages . put ( keyedMessage . getKey ( ) , keyedMessage ) ; } List < VehicleDataSink > deadSinks = new ArrayList <> ( ) ; for ( VehicleDataSink sink : mSinks ) { try { sink . receive ( message ) ; } catch ( DataSinkException e ) { Log . w ( TAG , this . getClass ( ) . getName ( ) + ": The sink " + sink + " exploded when we sent a new message " + "-- removing it from the pipeline: " + e ) ; deadSinks . add ( sink ) ; } } mMessagesReceived ++ ; for ( VehicleDataSink sink : deadSinks ) { removeSink ( sink ) ; } }
Accept new values from data sources and send it out to all registered sinks .