namespace BCG { using System.Collections.Generic; public class KeyValueToRelationalStream : ViewStream { private readonly ViewStream _kv_str; private readonly string _doc_id_col, _key_col, _val_col; public KeyValueToRelationalStream(ViewStream kv_str, string doc_id_col, string key_col, string val_col) { _kv_str = kv_str; _doc_id_col = doc_id_col; _key_col = key_col; _val_col = val_col; } protected override void myOnFirstSubscriber() { _kv_str.onChange += _on_str_update; } protected override void myOnLastSubscriber() { _kv_str.onChange -= _on_str_update; } private void _on_str_update(object dp_obj) { DataPacket dp = dp_obj as DataPacket; var rows = dp.newRows; var rv = BlueDataUtility.keyvalue_to_relational(rows, _doc_id_col, _key_col, _val_col); var dp_out = BlueDataUtility.build_dp_inc(rv, myLastValue as DataPacket, true); myFireEvent(dp_out); } } }