code: @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);
          }
      }
docs: Accept new values from data sources and send it out to all registered
  sinks.
  
  This method is required to implement the SourceCallback interface.
  
  If any data sink throws a DataSinkException when receiving data, it will
  be removed from the list of sinks.