diff --git a/src/DataStreamSuite/DataStream/Primitives/FileStreams/DumpLoaderStream.cs b/src/DataStreamSuite/DataStream/Primitives/FileStreams/DumpLoaderStream.cs deleted file mode 100644 index b9d7cea15..000000000 --- a/src/DataStreamSuite/DataStream/Primitives/FileStreams/DumpLoaderStream.cs +++ /dev/null @@ -1,49 +0,0 @@ -// using System; -// using System.Collections.Generic; -// using System.Text; -// using System.IO; - -// using System.Runtime.Serialization.Formatters.Binary; - -// namespace BCG -// { -// public class DumpLoaderStream : ViewStream, ISideEffectStream -// { -// //wtf -// //public class Exception : System.Exception -// //{public Exception(string msg) : base(msg) { }} - -// public DumpLoaderStream(string fileName, int secondsBetweenPolls) -// { -// mySetRealSource( -// new PollStream(() => new LoaderStream(fileName).rows, secondsBetweenPolls, -// ViewStreamDatabase.instance)); -// } - -// public DumpLoaderStream(string fileName) -// { -// if (fileName == null) -// throw new LoaderStream.Exception("File name cannot be null"); - -// for (int i = 0; i < 5; i++) -// try -// { -// using (FileStream fs = new FileStream(fileName, FileMode.Open, FileAccess.Read)) -// { -// IClientDeserializer deser = new ClientUniversalDeserializer(); -// byte[] the_ops; -// Array the_rows; -// deser.deserialize(new Stream[] { fs }, out the_ops, out the_rows); - -// mySetRealSource(new BlueRowCollectionStaticStream((IBlueRowCollection)the_rows.GetValue(0))); -// } -// break; -// } -// catch (IOException) -// { -// System.Threading.Thread.Sleep(50); -// } - -// } -// } -// } diff --git a/src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs b/src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs index 189be8954..b79e1a885 100644 --- a/src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs +++ b/src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs @@ -1,33 +1,7 @@ using BCG; namespace bts { - //[DataStreamUserRespresentation("saver")] - // public class SaverStream : ViewStream { - // private readonly ViewStream _vs; - // private readonly string _file_name; - // public SaverStream(ViewStream vs, string file_name) { - // _vs = vs; - // _file_name = file_name; - // } - - // protected override void myOnFirstSubscriber() { - // _vs.onChange += _handle_dp; - // } - - // protected override void myOnLastSubscriber() { - - // } - - // private void _handle_dp(object dp_obj) { - // DataPacket dp = dp_obj as DataPacket; - // byte[] serialized_update = null; - // var _serializer = new UniversalStreamIOSerializer(); - // serialized_update = _serializer.serialize(new bts.StreamIOMessage(new bts.StreamIOMessage.CreateMessage(dp.newRows))); - // StreamSerializationUtils.write_to_file(serialized_update, _file_name); - // myFireEvent(dp); - // } - // } - + // This is the other side of the SaverStream, its "loads" the stream and fires once [DataStreamUserRespresentation("loader")] public class LoaderStream : ViewStream { diff --git a/src/DataStreamSuite/DataStream/Primitives/FileStreams/PersistorStream.cs b/src/DataStreamSuite/DataStream/Primitives/FileStreams/PersistorStream.cs index fbb1022ff..1755bf76f 100644 --- a/src/DataStreamSuite/DataStream/Primitives/FileStreams/PersistorStream.cs +++ b/src/DataStreamSuite/DataStream/Primitives/FileStreams/PersistorStream.cs @@ -5,10 +5,11 @@ using System.Text; using System.IO; using Google.Protobuf; using System.Globalization; -using bts; +using BCG; -namespace BCG { - // Writes all the udpates to a stream in a file. +namespace bts { + // Writes all the udpates to a stream in a file or multiples files dependong on the rollover size (as opposed to Saver + // which just writes one update, as a create) // File naming: // token_epoch_yyyyMMdd_HHmmss // File location: directory(passed)\token\ @@ -94,10 +95,10 @@ namespace BCG { void _on_change(object new_value) { DataPacket dp = new_value as DataPacket; byte[] serialized_update = null; - bool _write_create = _serializer == null || _n_updates_in_current_file == 0; + bool first_write = _serializer == null || _n_updates_in_current_file == 0; _serializer = _serializer ?? new UniversalStreamIOSerializer(); - if (_write_create) { - // new files always starts with create + if (first_write) { + // first file always starts with create serialized_update = _serializer.serialize(new bts.StreamIOMessage(new bts.StreamIOMessage.CreateMessage(dp.newRows))); } else serialized_update = _serializer.serialize(new bts.StreamIOMessage(new bts.StreamIOMessage.MultipleRemoveAddMessage(dp.deletedRows, dp.addedRows))); @@ -105,8 +106,8 @@ namespace BCG { long utc_epoch = epoch_now(); //time, length of data, data - bw.Write(utc_epoch); - bw.Write(serialized_update.Length); + //bw.Write(utc_epoch); + //bw.Write(serialized_update.Length); bw.Write(serialized_update); _out_fs.Flush(); myFireEvent(dp); diff --git a/src/DataStreamSuite/DataStream/Primitives/FileStreams/PlaybackStream.cs b/src/DataStreamSuite/DataStream/Primitives/FileStreams/PlaybackStream.cs deleted file mode 100644 index 6e2674347..000000000 --- a/src/DataStreamSuite/DataStream/Primitives/FileStreams/PlaybackStream.cs +++ /dev/null @@ -1,190 +0,0 @@ -// using System; -// using System.Collections.Generic; -// using System.Linq; -// using System.Text; -// using System.IO; -// using System.Runtime.Serialization.Formatters.Binary; -// using BCG.Utilities.General; -// using bts.collections; -// using System.Reflection.Metadata; - - -// namespace BCG { - -// public class PlaybackStream : ViewStream, IWorker { -// private readonly PlaybackType _type; -// private BlueTable _current_table; -// private IClientDeserializer _deserilializer = new ClientUniversalDeserializer(); -// // these are the packets already downloaded from the current file -// private List _packets = new List(); -// //private IEnumerable _file_streams = new List(); -// private IEnumerable _file_names; -// private Tuple _current_fs = new Tuple(-1, null); -// private DateTime _playback_t0, _playback_t1; -// private bool _all_files_read = false; -// SnapshotData _snapshot_data = null; // only initialized when in snapshot mode - -// public PlaybackStream(string datetime_start, string datetime_end, string directory_name, string token) { -// _type = PlaybackType.Streaming; -// _playback_t0 = PersistorStream.parse_datetime(datetime_start); -// _playback_t1 = PersistorStream.parse_datetime(datetime_end); -// string token_directory_name = $"{directory_name}\\{token}"; -// bts.Log.Gen.info($"PlaybackStream t0: {_playback_t0} t1: {_playback_t1} dir: {token_directory_name}"); -// List files_in_range = new List(); -// if (Directory.Exists(token_directory_name)) { -// _file_names = -// Directory.GetFiles(token_directory_name).Where(f => { -// DateTime file_time = PersistorStream.parse_datetime_from_filename(f); -// return file_time > _playback_t0 && file_time < _playback_t1; -// }); -// } -// } - -// protected override void myOnFirstSubscriber() { -// FileStream fs = new FileStream(_snapshot_data.snapshot_file, FileMode.Open, FileAccess.Read); -// _snapshot_data.packets = StreamPersistorUtils.load_packets_from_file(fs); -// PlaybackPacket snapshot_packet = StreamPersistorUtils.find_packet_at_time(_snapshot_data.packets, _snapshot_data.snapshot_time); -// if (snapshot_packet != null) { -// foreach (PlaybackPacket p in _snapshot_data.packets) { -// byte[] ops; -// Array row_array; -// _deserilializer.deserialize(new Stream[] { new MemoryStream(p.data) }, out ops, out row_array); -// for (int i = 0; i < ops.Length; i++) { -// IBlueRowCollection[] rows_args = row_array.GetValue(i) as IBlueRowCollection[]; -// byte op = ops[i]; -// if (op == StreamOpCode.create) { -// IBlueRowCollection full_table = rows_args[0]; -// _current_table = new BlueTable(full_table); -// } else if (op == StreamOpCode.multipleRemoveAdd) { -// //IBlueRowCollection[] del_added = rows_args.GetValue(0) as IBlueRowCollection[]; -// IBlueRowCollection dels = rows_args.GetValue(0) as IBlueRowCollection; -// IBlueRowCollection adds = rows_args.GetValue(1) as IBlueRowCollection; -// BlueDataUtility.absorbChanges(_current_table, dels, adds); -// } else { -// bts.Log.assert(false, $"Oprator: {op} is not expected"); -// } -// if (p.idx == snapshot_packet.idx) { -// myFireEvent(new DataPacket(_current_table, _current_table, new BlueTable(_current_table.schema))); -// break; -// } -// } -// } -// } -// } - - -// private void _load_packets_from_current_file() { -// int curr_idx = _current_fs.Item1; -// FileStream curr_fs = _current_fs.Item2; -// _packets = new List(); -// _packets.AddRange(StreamPersistorUtils.load_packets_from_file(curr_fs)); -// } - -// private bool current_file_all_replayed { -// get { -// bool rv = true; -// _packets.ForEach(p => rv = rv && p.replayed); -// return rv; -// } -// } - -// protected override void myOnLastSubscriber() { -// var fs = _current_fs.Item2; -// if (fs != null) { -// fs.Close(); -// } -// ViewStreamDatabase.instance.removeWorker(this); -// } - -// // public event PlaybackStreamProgress onProgress; -// // public event Action onFileExhausted; -// private PlaybackPacket _locate_packet_to_replay() { -// foreach (var p in _packets) { -// // returns the first non replayed packet -// if (!p.replayed) -// return p; -// } -// //else -// { -// double secondsElapsed = new BinaryReader(myFileReader).ReadDouble(); -// myNextPacketTime = myPlaybackStart + new TimeSpan(0, 0, 0, 0, (int)(1000 * secondsElapsed / myPlaybackSpeed)); -// if (onProgress != null) -// onProgress(myFileReader.Position, myFileReader.Length); -// } -// } - -// #region IWorker Members - -// public void doWork() { -// if (_all_files_read) -// return; -// if (current_file_all_replayed) { -// _load_next_file(); -// } -// PlaybackPacket curr_packet = _locate_packet_to_replay(); -// if (curr_packet.idx == 0) { // first packet in the file, fire it rightaway -// byte[] ops; -// Array row_array; -// // what a stupid interface! -// _deserilializer.deserialize(new Stream[] { new MemoryStream(curr_packet.data) }, out ops, out row_array); -// _fire_packet(ops, row_array); -// curr_packet.replay_timestamp = PersistorStream.epoch_now(); -// } else { -// PlaybackPacket prev_packet = _get_previous_packet(curr_packet); -// // this is the time we need to fire (replay) this packet -// long time_to_fire = prev_packet.replay_timestamp + curr_packet.offset_from_previous; -// long now_epoch = PersistorStream.epoch_now(); -// // fire when we are past the -// if (now_epoch >= time_to_fire) { -// byte[] ops; -// Array row_array; -// _deserilializer.deserialize(new Stream[] { new MemoryStream(curr_packet.data) }, out ops, out row_array); -// _fire_packet(ops, row_array); -// curr_packet.replay_timestamp = PersistorStream.epoch_now(); -// } -// } -// } - -// private void _fire_packet(byte[] ops, Array row_array) { -// for (int i = 0; i < ops.Length; i++) { -// IBlueRowCollection[] rows_args = row_array.GetValue(i) as IBlueRowCollection[]; -// byte op = ops[i]; -// if (op == StreamOpCode.create) { -// IBlueRowCollection full_table = rows_args[0]; -// _current_table = new BlueTable(full_table); -// //_current_table.addRows(full_table); -// myFireEvent(new DataPacket(full_table, full_table, myLastValue == null ? new BlueTable(full_table.schema) : -// (myLastValue as DataPacket).newRows)); -// } else if (op == StreamOpCode.multipleRemoveAdd) { -// //IBlueRowCollection[] del_added = rows_args.GetValue(0) as IBlueRowCollection[]; -// IBlueRowCollection dels = rows_args.GetValue(0) as IBlueRowCollection; -// IBlueRowCollection adds = rows_args.GetValue(1) as IBlueRowCollection; -// BlueDataUtility.absorbChanges(_current_table, dels, adds); -// myFireEvent(new DataPacket(_current_table, adds, dels)); -// } else { -// bts.Log.assert(false, $"Oprator: {op} is not expected"); -// } - -// } -// } - -// private void _load_next_file() { -// var new_idx = _current_fs.Item1 + 1; -// if (new_idx >= _file_names.Count()) { -// _all_files_read = true; -// bts.Log.Gen.info("All files read"); -// return; -// } -// var prev_fs = _current_fs.Item2; -// if (prev_fs != null) -// prev_fs.Close(); -// var new_filename = _file_names.ElementAt(new_idx); -// bts.Log.Gen.info("PlaybackStream_load_next_file new_filename=: {new_filename}"); -// var new_fs = new FileStream(new_filename, FileMode.Open, FileAccess.Read); -// _current_fs = new Tuple(new_idx, new_fs); -// _load_packets_from_current_file(); -// } - -// #endregion -// } -// } diff --git a/src/DataStreamSuite/DataStream/Primitives/FileStreams/RecorderStream.cs b/src/DataStreamSuite/DataStream/Primitives/FileStreams/RecorderStream.cs deleted file mode 100644 index 1e7da4c49..000000000 --- a/src/DataStreamSuite/DataStream/Primitives/FileStreams/RecorderStream.cs +++ /dev/null @@ -1,68 +0,0 @@ -// using System; -// using System.Collections.Generic; -// using System.Linq; -// using System.Text; -// using System.IO; - -// namespace BCG -// { -// public class RecorderStream : ViewStream -// { -// private ViewStream mySource; -// private DateTime myStartTime; -// private FileStream myOutFile; -// private string myFileName; -// private IBlueDataSerializer mySerializer; - -// public RecorderStream(ViewStream src, string recordFile) -// { -// mySource = src; -// myFileName = recordFile; - -// myFileName = myFileName.Replace("%DATE%", DateTime.Now.ToString("yyyyMMdd_HHmmss")); -// } - -// protected override void myOnFirstSubscriber() -// { -// myStartTime = DateTime.Now; -// mySource.onChange += new ChangeDelegate(mySource_onChange); -// myOutFile = new FileStream(myFileName, FileMode.OpenOrCreate); -// } - -// protected override void myOnLastSubscriber() -// { -// mySource.onChange -= mySource_onChange; -// myOutFile.Close(); -// } - -// void mySource_onChange(object newValue) -// { -// DataPacket dp = (DataPacket)newValue; - -// BlueTable toWrite = new BlueTable(dp.schema); - -// if (mySerializer == null) -// { -// mySerializer = new UniversalSerializer(dp.schema); -// } -// foreach (IBlueRow r in dp.newRows) -// toWrite.addRow(r); - -// byte[] result; - -// if (myLastValue == null) -// mySerializer.serialize(StreamOpCode.create, new IBlueRowCollection[] { dp.newRows }, out result); -// else -// mySerializer.serialize(StreamOpCode.multipleRemoveAdd, new IBlueRowCollection[] { dp.deletedRows, dp.addedRows }, out result); - -// BinaryWriter bw = new BinaryWriter(myOutFile); -// //write down what time this happened -// bw.Write((DateTime.Now - myStartTime).TotalSeconds); -// bw.Write(result.Length); -// bw.Write(result); -// myOutFile.Flush(); - -// myFireEvent(dp); -// } -// } -// } diff --git a/src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs b/src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs index 41175e0fb..922b4022b 100644 --- a/src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs +++ b/src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs @@ -5,7 +5,8 @@ using System.IO; using BCG; namespace bts { - // this saves the total state in the filr provided in the cons + // this saves the updates as a Create on the same file, each update calls -save, which + // reqrites the whole data to the packet. The other side of this stream is the Loader [DataStreamUserRespresentation("saver")] public class SaverStream : ViewStream { private ViewStream _stream; diff --git a/src/DataStreamSuite/DataStreamPublisher/Configurations/MarketDataPublisher.json b/src/DataStreamSuite/DataStreamPublisher/Configurations/MarketDataPublisher.json index 100cb0e1d..2f7a35544 100644 --- a/src/DataStreamSuite/DataStreamPublisher/Configurations/MarketDataPublisher.json +++ b/src/DataStreamSuite/DataStreamPublisher/Configurations/MarketDataPublisher.json @@ -1,6 +1,6 @@ { "user": "bts", - "sparkd": "http://127.0.0.1:8000/config", + "sparkd": "http://192.168.1.80:8000/config", "pubsub": { "milliseconds_between_sub_info_updates": 1000, "milliseconds_between_time_slice_updates": 5000, @@ -105,5 +105,25 @@ } ] } + // { + // "name": "news_data", + // "streams": [ + // { + // "name": "benzinga_news", + // "expression": "benzingaNews(\"mongodb://localhost:27017\",\"local\",\"news_headlines\", \"30s\", \"5d\")", + // "is_on_websocket": true + // }, + // { + // "name": "benzinga_news_symbols", + // "expression": "benzingaNewsSymbols(benzinga_news)", + // "is_on_websocket": true + // }, + // { + // "name": "benzinga_news_symbol_state", + // "expression": "benzingaNewsSymbolState(benzinga_news)", + // "is_on_websocket": true + // } + // ] + // } ] } \ No newline at end of file diff --git a/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json b/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json index 67b031618..7154e2423 100644 --- a/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json +++ b/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json @@ -16,14 +16,14 @@ "publisher_routes": [ { "name": "default", - "server_address": "bluto://192.168.1.80:12345", + "server_address": "bts-cloud://192.168.1.80:12345", "tag": ".*" } ], "subscriber_routes": [ { "name": "default", - "server_address": "bluto://192.168.1.80:12345", + "server_address": "bts-cloud://192.168.1.80:12345", "tag": ".*" } ], diff --git a/src/DataStreamSuite/PubSub/Config/PubSubConfig.cs b/src/DataStreamSuite/PubSub/Config/PubSubConfig.cs index 605a8b823..f88da4e6f 100644 --- a/src/DataStreamSuite/PubSub/Config/PubSubConfig.cs +++ b/src/DataStreamSuite/PubSub/Config/PubSubConfig.cs @@ -117,10 +117,10 @@ public class PubSubConfig { WaitPeriodAfterGc = 1000, SimpleEchoServer = false, PublisherRoutes = new() { - new("bts-cloud://127.0.0.1:12345", ".*") + new("bts-cloud://192.168.1.80:12345", ".*") }, SubscriberRoutes = new() { - new("bts-cloud://127.0.0.1:12345", ".*") + new("bts-cloud://192.168.1.80:12345", ".*") } }; return rv;