[streaming] CoRecordReader and iterator added + ConnectedDataStream refactor