Function dataflow::source::handle_message

fn handle_message<S: SourceReader>(
    message: SourceMessage<S::Key, S::Value>, 
    bytes_read: &mut usize, 
    cap: &Capability<Timestamp>, 
    output: &mut OutputHandle<'_, Timestamp, Result<SourceOutput<S::Key, S::Value>, String>, Tee<Timestamp, Result<SourceOutput<S::Key, S::Value>, String>>>, 
    metric_updates: &mut HashMap<PartitionId, (MzOffset, Timestamp, i64)>, 
    timer: &Instant, 
    timestamp_bindings: &TimestampBindingRc
) -> (SourceStatus, MessageProcessing)

Take message and assign it the appropriate timestamps and push it into the dataflow layer, if possible.

TODO: This function is a bit of a mess rn but hopefully this function makes the existing mess more obvious and points towards ways to improve it.