Function subscribe 

fn subscribe<'scope>(
    sinked_collection: VecCollection<'scope, Timestamp, Row, Diff>,
    err_collection: VecCollection<'scope, Timestamp, DataflowError, Diff>,
    sink_id: GlobalId,
    with_snapshot: bool,
    as_of: Antichain<Timestamp>,
    up_to: Antichain<Timestamp>,
    subscribe_protocol_handle: Rc<RefCell<Option<SubscribeProtocol>>>,
)