diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json
index 7b6c0b4ee..1a9d66335 100644
--- a/.devcontainer/devcontainer.json
+++ b/.devcontainer/devcontainer.json
@@ -27,7 +27,7 @@
"-p",
"8003:8003",
"-p",
- "12345:12345",
+ "12345:12345"
],
"mounts": [
"source=${localEnv:HOME}/.bts/devcontainer-mounts/${devcontainerId},target=/home/bts,type=bind"
diff --git a/.vscode/launch.json b/.vscode/launch.json
index 2bfc14f64..73b25d8e8 100644
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -30,7 +30,7 @@
"name": "DataStreamQueryAnalyzer",
"type": "coreclr",
"request": "launch",
- "preLaunchTask": "Build DataStream (Debug)",
+ //"preLaunchTask": "Build DataStreamQueryAnalyzer (Debug)",
"program": "",
"windows": {
"program": "DataStreamQueryAnalyzer.exe",
@@ -143,4 +143,4 @@
"default": "http://localhost:8000/config",
},
]
-}
+}
\ No newline at end of file
diff --git a/src/DataStreamSuite/DataStream/Primitives/Distinct/DistinctRowStream.cs b/src/DataStreamSuite/DataStream/Primitives/Distinct/DistinctRowStream.cs
index 8184b378a..a8d76a948 100644
--- a/src/DataStreamSuite/DataStream/Primitives/Distinct/DistinctRowStream.cs
+++ b/src/DataStreamSuite/DataStream/Primitives/Distinct/DistinctRowStream.cs
@@ -1,23 +1,24 @@
+using bts;
+
using System;
using System.Collections;
-namespace BCG
-{
+namespace BCG {
///
/// Provides a stream that with distinct rows with respect to the columns
/// the user specifies in the argument
///
- [DataStreamUserRespresentation ("distinct")]
- public class DistinctRowStream : ViewStream
- {
- public DistinctRowStream (ViewStream stream, params string[] cols)
- {
- ArrayList groupByCols = new ArrayList ();
- foreach (string col in cols)
- foreach (string c in col.Split (' '))
- groupByCols.Add (c);
- mySetRealSource (new GroupByStream (stream,
- new DistinctRowEvaluator (groupByCols.ToArray (typeof(string)) as string[]), false));
+ [DataStreamUserRespresentation("distinct")]
+ public class DistinctRowStream : ViewStream {
+ public DistinctRowStream(ViewStream stream, params string[] cols) {
+ var g_by_eval = new DistinctRowEvaluator(cols);
+ ViewStream r = new GroupByStream(stream, g_by_eval);
+ var proj_exp = String.Empty;
+ foreach (var c in cols)
+ proj_exp += $"{c} ";
+ proj_exp = proj_exp.Trim();
+ r = new NewProjectStream(r, proj_exp);
+ mySetRealSource(r);
}
}
}
diff --git a/src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs b/src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs
index 8cb3ebc1c..189be8954 100644
--- a/src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs
+++ b/src/DataStreamSuite/DataStream/Primitives/FileStreams/LoaderStream.cs
@@ -1,6 +1,33 @@
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);
+ // }
+ // }
+
[DataStreamUserRespresentation("loader")]
public class LoaderStream : ViewStream {
diff --git a/src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs b/src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs
index 1af59181b..41175e0fb 100644
--- a/src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs
+++ b/src/DataStreamSuite/DataStream/Primitives/FileStreams/SaverStream.cs
@@ -31,7 +31,6 @@ namespace bts {
StreamIOMessage.CreateMessage cr_msg = new StreamIOMessage.CreateMessage(dp.newRows);
byte[] str_bytes = _serializer.serialize(new StreamIOMessage(cr_msg));
StreamSerializationUtils.write_to_file(str_bytes, _filename);
-
}
}
}
diff --git a/src/DataStreamSuite/DataStream/Primitives/Misc/PubSubTestStream.cs b/src/DataStreamSuite/DataStream/Primitives/Misc/PubSubTestStream.cs
index 36afb6811..0225ecb7b 100644
--- a/src/DataStreamSuite/DataStream/Primitives/Misc/PubSubTestStream.cs
+++ b/src/DataStreamSuite/DataStream/Primitives/Misc/PubSubTestStream.cs
@@ -8,26 +8,33 @@ using System;
using System.IO;
using System.Linq;
-namespace bts {
- public class PubSubTestingStream : ViewStream, IWorker {
- public class PayloadMessage {
+namespace bts
+{
+ public class PubSubTestingStream : ViewStream, IWorker
+ {
+ public class PayloadMessage
+ {
public int msg_id = 0;
public string guid;
public Int64 travel_epoch_ms = 0;
public byte[] payload = null;
- public PayloadMessage(int msg_id_args, Int64 send_epoch_args, int payload_size, byte payload_byte) {
+ public PayloadMessage(int msg_id_args, Int64 send_epoch_args, int payload_size, byte payload_byte)
+ {
msg_id = msg_id_args;
travel_epoch_ms = send_epoch_args;
payload = new byte[payload_size];
- for (int i = 0; i < payload_size; i++) {
+ for (int i = 0; i < payload_size; i++)
+ {
payload[i] = payload_byte;
}
}
private PayloadMessage() { }
public PayloadMessage(int msg_id, int payload_size, byte payload_byte) :
- this(msg_id, PubSubTestingStream.epoch_now(), payload_size, payload_byte) { }
+ this(msg_id, PubSubTestingStream.epoch_now(), payload_size, payload_byte)
+ { }
- public byte[] serialize() {
+ public byte[] serialize()
+ {
var str = new MemoryStream();
BinaryWriter bw = new BinaryWriter(str);
bw.Write(msg_id);
@@ -40,7 +47,8 @@ namespace bts {
return bytes;
}
- public static PayloadMessage deserialize(byte[] ser_bytes) {
+ public static PayloadMessage deserialize(byte[] ser_bytes)
+ {
var str = new MemoryStream(ser_bytes);
BinaryReader br = new BinaryReader(str);
int msg_id = br.ReadInt32();
@@ -48,7 +56,8 @@ namespace bts {
Int64 epoch = br.ReadInt64();
int len = br.ReadInt32();
var payload = br.ReadBytes(len);
- return new PayloadMessage() {
+ return new PayloadMessage()
+ {
msg_id = msg_id,
guid = guid,
travel_epoch_ms = epoch,
@@ -73,7 +82,8 @@ namespace bts {
private readonly int _payload_sz;
private const byte _payload_byte = (byte)'A';
private readonly string _tag_name;
- public PubSubTestingStream(int publish_period_msecs, int max_pubs, int payload_size, string tag_name, ViewStreamDatabase vsd) {
+ public PubSubTestingStream(int publish_period_msecs, int max_pubs, int payload_size, string tag_name, ViewStreamDatabase vsd)
+ {
_vsd = vsd;
_payload_sz = payload_size;
var schema = BlueDataUtility.form_schema("PubSubTest",
@@ -96,43 +106,53 @@ namespace bts {
public PubSubTestingStream(int publish_period_seconds, int max_pubs, int msg_size, string tag_name)
: this(publish_period_seconds, max_pubs, msg_size, tag_name, ViewStreamDatabase.instance) { }
- protected override void myOnFirstSubscriber() {
+ protected override void myOnFirstSubscriber()
+ {
_vsd.addWorker(this);
System.Threading.Thread th = new System.Threading.Thread(_subscribe_thread_func);
th.IsBackground = true;
th.Start();
}
- protected override void myOnLastSubscriber() {
+ protected override void myOnLastSubscriber()
+ {
_vsd.removeWorker(this);
_continue_sub_thread = false;
}
- void _log_updates(IEnumerable updates) {
+ void _log_updates(IEnumerable updates)
+ {
var idx = 0;
- foreach (var msg in updates) {
+ foreach (var msg in updates)
+ {
bts.Log.Gen.info($"PubSubTesStream._subscribe_thread_func idx: {idx} msg: {msg}");
idx++;
}
}
- private void _subscribe_thread_func() {
+ private void _subscribe_thread_func()
+ {
// subscription is done in another thread
SubscriberCore sub_core = new SubscriberCore();
sub_core.subscribe(_tag_name);
- while (_continue_sub_thread) {
+ while (_continue_sub_thread)
+ {
// TODO recv_id
var updates = sub_core.get_updates(TimeSpan.FromMilliseconds(250));
if (updates == null || updates.Count() == 0)
continue;
_log_updates(updates);
_n_updates += updates.Count();
- if (updates.Count() > 0) {
+ if (updates.Count() > 0)
+ {
var server_msg = updates.Last();
- if (server_msg.data != null && server_msg.data.Length > 0) {
+ if (server_msg.data != null && server_msg.data.Length > 0)
+ {
_last_recvd_msg = PayloadMessage.deserialize(server_msg.data);
_last_recvd_msg.travel_epoch_ms = epoch_now();
- } else {
+ }
+ else
+ {
// why are we getting empty or null data ? (server is sending data regardless?)
}
_fire_dirty = true;
@@ -140,29 +160,36 @@ namespace bts {
}
}
- private void _publish_next_packet() {
- if (_last_sent_msg == null) {
- _last_sent_msg = new PayloadMessage(0, _payload_sz, _payload_byte) {
+ private void _publish_next_packet()
+ {
+ if (_last_sent_msg == null)
+ {
+ _last_sent_msg = new PayloadMessage(0, _payload_sz, _payload_byte)
+ {
guid = Guid.NewGuid().ToString()
};
- } else {
+ }
+ else
+ {
_last_sent_msg.msg_id++;
_last_sent_msg.guid = Guid.NewGuid().ToString();
_last_sent_msg.travel_epoch_ms = epoch_now();
}
var ser_bytes = _last_sent_msg.serialize();
_n_pubs++;
- _pub_core.publish(String.Empty, _tag_name, ser_bytes);
+ _pub_core.publish("Blob", _tag_name, ser_bytes);
_fire_dirty = true;
}
- private void _fire_dp() {
+ private void _fire_dp()
+ {
object last_sent_id = _last_sent_msg == null ? (object)DBNull.Value : (object)_last_sent_msg.msg_id;
object last_recv_id = _last_recvd_msg == null ? (object)DBNull.Value : (object)_last_recvd_msg.msg_id;
object echo_time = DBNull.Value;
object last_sent_guid = _last_sent_msg == null ? (object)DBNull.Value : (object)_last_sent_msg.guid;
object last_recv_guid = _last_recvd_msg == null ? (object)DBNull.Value : (object)_last_recvd_msg.guid;
- if (_last_sent_msg != null && _last_recvd_msg != null) {
+ if (_last_sent_msg != null && _last_recvd_msg != null)
+ {
echo_time = _last_recvd_msg.travel_epoch_ms - _last_sent_msg.travel_epoch_ms;
}
var match = last_sent_id.Equals(last_recv_id) && last_sent_guid.Equals(last_recv_guid);
@@ -174,21 +201,25 @@ namespace bts {
myFireEvent(dp);
}
- public void doWork() {
+ public void doWork()
+ {
if (DateTime.Now - _last_publish_time > _publish_period)
- if (_n_pubs <= _max_pubs) {
+ if (_n_pubs <= _max_pubs)
+ {
_publish_next_packet();
_last_publish_time = DateTime.Now;
}
// publish
// TODO: send id
- if (_fire_dirty) {
+ if (_fire_dirty)
+ {
_fire_dirty = false;
_fire_dp();
}
}
- public static long epoch_now() {
+ public static long epoch_now()
+ {
return DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
}
diff --git a/src/DataStreamSuite/DataStream/Primitives/Misc/PubsubEchoTestStream.cs b/src/DataStreamSuite/DataStream/Primitives/Misc/PubsubEchoTestStream.cs
new file mode 100644
index 000000000..b37570af9
--- /dev/null
+++ b/src/DataStreamSuite/DataStream/Primitives/Misc/PubsubEchoTestStream.cs
@@ -0,0 +1,223 @@
+using BCG;
+using BCG.Utilities.General;
+using System.Collections.Generic;
+
+
+using System;
+using System.IO;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using bts.api;
+
+namespace bts
+{
+ [DataStreamUserRespresentation("pss_echo")]
+ public class PubsubEchoTestStream : ViewStream, IWorker
+ {
+ public class PayloadMessage
+ {
+ public readonly string msg_id;
+ public readonly Int64 timestamp = 0;
+ public readonly byte[] payload = null;
+
+ public static byte[] create_data(int payload_size, byte payload_byte)
+ {
+ var rv = new byte[payload_size];
+ for (int i = 0; i < payload_size; i++)
+ {
+ rv[i] = payload_byte;
+ }
+ return rv;
+ }
+
+ // public PayloadMessage(string guid_, Int64 timestamp_, int payload_size, byte payload_byte)
+ // {
+ // guid = guid_;
+ // timestamp = timestamp_;
+ // payload = new byte[payload_size];
+ // for (int i = 0; i < payload_size; i++)
+ // {
+ // payload[i] = payload_byte;
+ // }
+ // }
+ public PayloadMessage(string guid_, Int64 timestamp_, byte[] payload_)
+ {
+ msg_id = guid_;
+ timestamp = timestamp_;
+ payload = payload_;
+ }
+
+ private PayloadMessage() { }
+ public byte[] serialize()
+ {
+ var str = new MemoryStream();
+ BinaryWriter bw = new BinaryWriter(str);
+ bw.Write(msg_id);
+ bw.Write(timestamp);
+ bw.Write(payload.Length);
+ bw.Write(payload);
+ str.Position = 0;
+ var bytes = str.ToArray();
+ return bytes;
+ }
+
+ public static PayloadMessage deserialize(byte[] ser_bytes)
+ {
+ var str = new MemoryStream(ser_bytes);
+ BinaryReader br = new BinaryReader(str);
+ string guid = br.ReadString();
+ Int64 timestamp_ = br.ReadInt64();
+ int len = br.ReadInt32();
+ var payload = br.ReadBytes(len);
+ return new PayloadMessage(guid, timestamp_, payload);
+ }
+ }
+ //public void
+ private readonly int _msg_size;
+ private readonly TimeSpan _publish_period;
+ private DateTime _last_publish_time = DateTime.MinValue;
+ private PublisherCore _pub_core;
+ private ISubscriberCore _sub_core;
+ private BlueTable _table;
+ private int _n_pubs = 0;
+ private readonly int _max_pubs;
+ private int _n_updates = 0;
+ private ViewStreamDatabase _vsd;
+ private PayloadMessage _last_sent_msg, _last_recvd_msg;
+ private const byte _payload_byte = (byte)'A';
+ private readonly string _tag_name;
+ private readonly int _payload_sz;
+ private readonly IPubSubRouter _router;
+ private enum IOState
+ {
+ SEND = 0,
+ RECV = 1
+ };
+ IOState _state = IOState.SEND;
+
+ public PubsubEchoTestStream(string server_url, TimeSpan publish_period, int payload_size, string tag_name, ViewStreamDatabase vsd)
+ {
+ var route = new PubSubRoute(server_url, ".*");
+ _router = new PubSubRouter(route);
+ _vsd = vsd;
+ _payload_sz = payload_size;
+ var schema = BlueDataUtility.form_schema("PubSubTest",
+ Tuple.Create("id", TypeCode.Int32),
+ Tuple.Create("sent_id", TypeCode.String),
+ Tuple.Create("sent_time", TypeCode.Int64),
+ Tuple.Create("recv_id", TypeCode.String),
+ Tuple.Create("recv_time", TypeCode.Int64),
+ Tuple.Create("match", TypeCode.Boolean),
+ Tuple.Create("echo_time", TypeCode.Int32));
+ _table = new BlueTable(schema);
+ _publish_period = publish_period;
+ _tag_name = tag_name;
+ }
+
+ public PubsubEchoTestStream(string server_url, int publish_period_msecs, int payload_size, string tag_name)
+ : this(server_url, TimeSpan.FromMilliseconds(publish_period_msecs), payload_size, tag_name, ViewStreamDatabase.instance) { }
+
+ protected override void myOnFirstSubscriber()
+ {
+ _pub_core = new PublisherCore(_router);
+ _sub_core = new SubscriberCore(_router);
+ _sub_core.subscribe(_tag_name);
+ _vsd.addWorker(this);
+
+ }
+
+ protected override void myOnLastSubscriber()
+ {
+ _vsd.removeWorker(this);
+ _sub_core.unsubscribe(_tag_name);
+ }
+
+ private void _fire_dp_after_send()
+ {
+ // as soon as the data is published, its reflected in the stream state
+ Log.assert(_last_sent_msg != null);
+ BlueRow r = new BlueRow(_table.schema, _table.Count, _last_sent_msg.msg_id, _last_sent_msg.timestamp, DBNull.Value, DBNull.Value,
+ DBNull.Value, DBNull.Value);
+ _table.addRow(r);
+ var dp_out = BlueDataUtility.build_dp_inc(_table, myLastValue as DataPacket);
+ myFireEvent(dp_out);
+ }
+
+ private void _fire_dp_after_recv()
+ {
+ Log.assert(_last_recvd_msg != null);
+ // update the last publish row
+ int id = _table.Count - 1;
+ IBlueRow last_row = _table[id];
+ _table.deleteRow(last_row);
+ var recv_ts = epoch_now();
+ BlueRow nr = new BlueRow(_table.schema, id, _last_sent_msg.msg_id, _last_sent_msg.timestamp, _last_recvd_msg.msg_id, recv_ts,
+ String.Compare(_last_recvd_msg.msg_id, _last_sent_msg.msg_id) == 0, recv_ts - _last_recvd_msg.timestamp);
+ _table.addRow(nr);
+ var dp_out = BlueDataUtility.build_dp_inc(_table, myLastValue as DataPacket);
+ myFireEvent(dp_out);
+ }
+
+ private string _get_next_id()
+ {
+ return _table.Count.ToString();
+ }
+
+ private void _publish_next_packet()
+ {
+ var bytes = PayloadMessage.create_data(_payload_sz, _payload_byte);
+ _last_sent_msg = new PayloadMessage(_get_next_id(), epoch_now(), bytes);
+ var ser_bytes = _last_sent_msg.serialize();
+ _n_pubs++;
+ _pub_core.publish("Blob", _tag_name, ser_bytes);
+ _last_publish_time = DateTime.Now;
+ }
+
+ public void doWork()
+ {
+ Func, PayloadMessage> find_packet = (updates) =>
+ {
+ foreach (var u in updates)
+ {
+ var recv_msg = PayloadMessage.deserialize(u.data);
+ if (recv_msg.msg_id == _last_sent_msg.msg_id)
+ {
+ return recv_msg;
+ }
+ }
+ return null;
+ };
+ switch (_state)
+ {
+ case IOState.SEND:
+ if (DateTime.Now - _last_publish_time >= _publish_period)
+ {
+ _publish_next_packet();
+ _last_publish_time = DateTime.Now;
+ _fire_dp_after_send();
+ _state = IOState.RECV;
+ return;
+ }
+ break;
+ case IOState.RECV:
+ var updates = _sub_core.get_updates(TimeSpan.Zero);
+ updates = updates.Where(u => u.tag == _tag_name);
+ var recv_msg = find_packet(updates);
+ if (recv_msg != null)
+ {
+ _last_recvd_msg = recv_msg;
+ _fire_dp_after_recv();
+ _state = IOState.SEND;
+ break;
+ }
+ break;
+ }
+ }
+
+ public static long epoch_now()
+ {
+ return DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+ }
+
+ }
+}
\ No newline at end of file
diff --git a/src/DataStreamSuite/DataStream/Primitives/MongoDB/MongoDBStream.cs b/src/DataStreamSuite/DataStream/Primitives/MongoDB/MongoDBStream.cs
index 00dcd567e..3c92f13de 100644
--- a/src/DataStreamSuite/DataStream/Primitives/MongoDB/MongoDBStream.cs
+++ b/src/DataStreamSuite/DataStream/Primitives/MongoDB/MongoDBStream.cs
@@ -7,122 +7,127 @@ using MongoDB.Driver;
using System;
using System.Collections.Generic;
using System.Diagnostics;
+using System.Threading.Tasks;
+using System.Threading;
namespace bts {
- [DataStreamUserRespresentation("mongodb_sql")]
- public class MongoDBSQLStream : ViewStream, IWorker {
- private ViewStreamDatabase _vsd;
- private readonly string _connection_string, _db_name, _collection_name;
- private MongoClient _client;
- // make this a sorted table
- private BlueTable _table;
+ // [DataStreamUserRespresentation("mongodb_sql")]
+ // public class MongoDBSQLStream : ViewStream, IWorker {
+ // private ViewStreamDatabase _vsd;
+ // private readonly string _connection_string, _db_name, _collection_name;
+ // private MongoClient _client;
+ // // make this a sorted table
+ // private BlueTable _table;
- // Access the database
- private IMongoDatabase _database;
- private IDictionary _id_map = new Dictionary();
- // mongodb://192.168.1.80:27017
+ // // Access the database
+ // private IMongoDatabase _database;
+ // private IDictionary _id_map = new Dictionary();
+ // // mongodb://192.168.1.80:27017
- public MongoDBSQLStream(string connection_string, string db_name, string collection_name)
- : this(connection_string, db_name, collection_name, ViewStreamDatabase.instance) {
+ // public MongoDBSQLStream(string connection_string, string db_name, string collection_name)
+ // : this(connection_string, db_name, collection_name, ViewStreamDatabase.instance) {
- }
- public MongoDBSQLStream(string connection_string, string db_name, string collection_name, ViewStreamDatabase vsd) {
- bts.Log.Gen.info($"MongoDBStream constructed connection_string: {connection_string} " +
- $"db_name: {db_name}");
- _connection_string = connection_string;
- _db_name = db_name;
- _collection_name = collection_name;
- _vsd = vsd;
- _client = new MongoClient(_connection_string);
- _database = _client.GetDatabase(_db_name);
- }
+ // }
+ // public MongoDBSQLStream(string connection_string, string db_name, string collection_name, ViewStreamDatabase vsd) {
+ // bts.Log.Gen.info($"MongoDBStream constructed connection_string: {connection_string} " +
+ // $"db_name: {db_name}");
+ // _connection_string = connection_string;
+ // _db_name = db_name;
+ // _collection_name = collection_name;
+ // _vsd = vsd;
+ // _client = new MongoClient(_connection_string);
+ // _database = _client.GetDatabase(_db_name);
+ // }
- protected override void myOnFirstSubscriber() {
- MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(10));
- }
+ // protected override void myOnFirstSubscriber() {
+ // MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(10));
+ // }
- protected override void myOnLastSubscriber() {
- MasterWorker.getInstance(false).removeWorker(this);
- }
+ // protected override void myOnLastSubscriber() {
+ // MasterWorker.getInstance(false).removeWorker(this);
+ // }
- public void doWork() {
- _read_mogodb();
- }
+ // public void doWork() {
+ // _read_mogodb();
+ // }
- private void _read_mogodb() {
- var timer = Stopwatch.StartNew();
- // Access the collection
- IMongoCollection collection = _database.GetCollection(name: _collection_name);
+ // private void _read_mogodb() {
+ // var timer = Stopwatch.StartNew();
+ // // Access the collection
+ // IMongoCollection collection = _database.GetCollection(name: _collection_name);
- // Fetch all documents from the collection
- var documents = collection.Find(Builders.Filter.Empty).ToCursor();
- _table = _table == null ? null : new BlueTable(_table.schema);
- // Iterate through the documents and print values
- foreach (var document in documents.ToEnumerable()) {
- bts.Log.Gen.info($"document: {document} json: {document.ToJson()}");
- if (_table == null) {
- _init_table(document);
- }
- _add_row(document);
- }
- timer.Stop();
- Log.Gen.info($"MongoDBStream time to read {_collection_name} {timer.ElapsedMilliseconds}ms");
- timer.Restart();
- var dp = BlueDataUtility.build_dp_inc(_table, myLastValue == null ? null : myLastValue as DataPacket, true);
- timer.Stop();
- Log.Gen.info($"MongoDBStream time to compare tables :{timer.ElapsedMilliseconds}ms");
- if (dp.addedRows.Count == 0 && dp.deletedRows.Count == 0)
- return;
- myFireEvent(dp);
- }
+ // // Fetch all documents from the collection
+ // var documents = collection.Find(Builders.Filter.Empty).ToCursor();
+ // _table = _table == null ? null : new BlueTable(_table.schema);
+ // // Iterate through the documents and print values
+ // foreach (var document in documents.ToEnumerable()) {
+ // bts.Log.Gen.info($"document: {document} json: {document.ToJson()}");
+ // if (_table == null) {
+ // _init_table(document);
+ // }
+ // _add_row(document);
+ // }
+ // timer.Stop();
+ // Log.Gen.info($"MongoDBStream time to read {_collection_name} {timer.ElapsedMilliseconds}ms");
+ // timer.Restart();
+ // var dp = BlueDataUtility.build_dp_inc(_table, myLastValue == null ? null : myLastValue as DataPacket, true);
+ // timer.Stop();
+ // Log.Gen.info($"MongoDBStream time to compare tables :{timer.ElapsedMilliseconds}ms");
+ // if (dp.addedRows.Count == 0 && dp.deletedRows.Count == 0)
+ // return;
+ // myFireEvent(dp);
+ // }
- private void _init_table(BsonDocument doc_row) {
- bts.Log.assert(_table == null);
- List cis = new List();
- cis.Add((new ColumnInfo("id", 0, typeof(Int32))));
- foreach (var n in doc_row.Names) {
- cis.Add(new ColumnInfo(n, cis.Count, typeof(string)));
- }
- IBlueTableSchema schema = new BlueTableSchema(cis.ToArray(), _collection_name);
- _table = new BlueTable(schema);
- }
+ // private void _init_table(BsonDocument doc_row) {
+ // bts.Log.assert(_table == null);
+ // List cis = new List();
+ // cis.Add((new ColumnInfo("id", 0, typeof(Int32))));
+ // foreach (var n in doc_row.Names) {
+ // cis.Add(new ColumnInfo(n, cis.Count, typeof(string)));
+ // }
+ // IBlueTableSchema schema = new BlueTableSchema(cis.ToArray(), _collection_name);
+ // _table = new BlueTable(schema);
+ // }
- private void _add_row(BsonDocument doc) {
- var key_val = doc["_id"];
- var key = key_val.ToString();
- int id;
- if (!_id_map.TryGetValue(key, out id)) {
- id = _id_map.Count;
- _id_map.Add(key, id);
- }
- MutableBlueRow nr = new MutableBlueRow(_table.schema);
- nr.setValue("id", id);
- foreach (var n in doc.Names) {
- var val = doc[n].ToString();
- nr.setValue(n, val);
- }
- _table.addRow(nr);
- }
- }
+ // private void _add_row(BsonDocument doc) {
+ // var key_val = doc["_id"];
+ // var key = key_val.ToString();
+ // int id;
+ // if (!_id_map.TryGetValue(key, out id)) {
+ // id = _id_map.Count;
+ // _id_map.Add(key, id);
+ // }
+ // MutableBlueRow nr = new MutableBlueRow(_table.schema);
+ // nr.setValue("id", id);
+ // foreach (var n in doc.Names) {
+ // var val = doc[n].ToString();
+ // nr.setValue(n, val);
+ // }
+ // _table.addRow(nr);
+ // }
+ // }
[DataStreamUserRespresentation("mongodb")]
public class MongoDBStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
private readonly string _connection_string, _db_name, _collection_name;
+ private readonly TimeSpan _poll_period;
private MongoClient _client;
private BlueTable _table;
+ private SyncQueue _table_queue = new SyncQueue();
+ private BlockingQueue _work_queue = new BlockingQueue();
// Access the database
private IMongoDatabase _database;
private IDictionary _id_map = new Dictionary();
// mongodb://192.168.1.80:27017
- public MongoDBStream(string connection_string, string db_name, string collection_name)
- : this(connection_string, db_name, collection_name, ViewStreamDatabase.instance) {
+ public MongoDBStream(string connection_string, string db_name, string collection_name, string poll_period)
+ : this(connection_string, db_name, collection_name, TimeBasedFilter.parse_duration(poll_period), ViewStreamDatabase.instance) {
}
- public MongoDBStream(string connection_string, string db_name, string collection_name, ViewStreamDatabase vsd) {
+ public MongoDBStream(string connection_string, string db_name, string collection_name, TimeSpan poll_period, ViewStreamDatabase vsd) {
bts.Log.Gen.info($"MongoDBStream constructed connection_string: {connection_string} " +
$"db_name: {db_name}");
_connection_string = connection_string;
@@ -130,19 +135,30 @@ namespace bts {
_collection_name = collection_name;
_vsd = vsd;
_client = new MongoClient(_connection_string);
+ _poll_period = poll_period;
_database = _client.GetDatabase(_db_name);
}
protected override void myOnFirstSubscriber() {
- MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(10));
+ _vsd.addWorker(this);
+ //MasterWorker.getInstance(false).addWorker(this, _poll_period);
}
protected override void myOnLastSubscriber() {
- MasterWorker.getInstance(false).removeWorker(this);
+ _vsd.removeWorker(this);
+ //MasterWorker.getInstance(false).removeWorker(this);
}
+ private DateTime _last_poll_time = DateTime.MinValue;
public void doWork() {
- _read_mogodb();
+ var now = DateTime.Now;
+ if (now - _last_poll_time > _poll_period) {
+ ThreadPool.QueueUserWorkItem(_ => _read_mogodb());
+ _last_poll_time = DateTime.Now;
+ }
+ var dp = _table_queue.dequeue(0);
+ if (dp != null)
+ myFireEvent(dp);
}
private void _read_mogodb() {
@@ -166,10 +182,11 @@ namespace bts {
timer.Restart();
var dp = BlueDataUtility.build_dp_inc(_table, myLastValue == null ? null : myLastValue as DataPacket, true);
timer.Stop();
- Log.Gen.info($"MongoDBStream time to compare tables: {timer.ElapsedMilliseconds}ms");;
+ Log.Gen.info($"MongoDBStream time to compare tables: {timer.ElapsedMilliseconds}ms"); ;
if (dp.addedRows.Count == 0 && dp.deletedRows.Count == 0)
return;
- myFireEvent(dp);
+ _table_queue.enqueue(dp);
+ //myFireEvent(dp);
}
private void _init_table(BsonDocument doc_row) {
@@ -208,46 +225,46 @@ namespace bts {
}
}
- [DataStreamUserRespresentation("mongodb_flat")]
- public class MongoDBFlatStream : ViewStream {
- public MongoDBFlatStream(string connection_string, string db_name, string collection_name, string timestamp_col, string ttl_duration)
- :
- this(ViewStreamDatabase.instance, connection_string, db_name, collection_name, timestamp_col, TimeBasedFilter.parse_duration(ttl_duration)) { }
+ // [DataStreamUserRespresentation("mongodb_flat")]
+ // public class MongoDBFlatStream : ViewStream {
+ // public MongoDBFlatStream(string connection_string, string db_name, string collection_name, string timestamp_col, string ttl_duration)
+ // :
+ // this(ViewStreamDatabase.instance, connection_string, db_name, collection_name, timestamp_col, TimeBasedFilter.parse_duration(ttl_duration)) { }
- public MongoDBFlatStream(ViewStreamDatabase vsd, string connection_string, string db_name, string collection_name, string timestamp_col, TimeSpan ttl_duration) {
- ViewStream rs = new MongoDBStream(connection_string, db_name, collection_name, vsd);
- rs = new KeyValueToRelationalStream(rs, "doc_id", "key", "val");
- rs = new DateStandardizationStream(rs, timestamp_col, DateConversion.ISO8601);
- //rs = new TimeBasedFilterStream(vsd, rs, timestamp_col, ttl_duration);
- mySetRealSource(rs);
- }
- }
+ // public MongoDBFlatStream(ViewStreamDatabase vsd, string connection_string, string db_name, string collection_name, string timestamp_col, TimeSpan ttl_duration) {
+ // ViewStream rs = new MongoDBStream(connection_string, db_name, collection_name, vsd);
+ // rs = new KeyValueToRelationalStream(rs, "doc_id", "key", "val");
+ // rs = new DateStandardizationStream(rs, timestamp_col, DateConversion.ISO8601);
+ // //rs = new TimeBasedFilterStream(vsd, rs, timestamp_col, ttl_duration);
+ // mySetRealSource(rs);
+ // }
+ // }
- [DataStreamUserRespresentation("newsreplay")]
- public class NewsReplayStream : ViewStream {
- public NewsReplayStream(string connection_string, string db_name, string collection_name,
- int buffer_sz, string fire_period) :
- this(ViewStreamDatabase.instance, connection_string, db_name, collection_name, buffer_sz, TimeBasedFilter.parse_duration(fire_period)) { }
+ // [DataStreamUserRespresentation("newsreplay")]
+ // public class NewsReplayStream : ViewStream {
+ // public NewsReplayStream(string connection_string, string db_name, string collection_name,
+ // int buffer_sz, string fire_period) :
+ // this(ViewStreamDatabase.instance, connection_string, db_name, collection_name, buffer_sz, TimeBasedFilter.parse_duration(fire_period)) { }
- public NewsReplayStream(ViewStreamDatabase vsd, string connection_string, string db_name, string collection_name,
- int buffer_sz, TimeSpan fire_period) {
- ViewStream r;
+ // public NewsReplayStream(ViewStreamDatabase vsd, string connection_string, string db_name, string collection_name,
+ // int buffer_sz, TimeSpan fire_period) {
+ // ViewStream r;
- r = new MongoDBStream(connection_string, db_name, collection_name, vsd);
- r = new KeyValueToRelationalStream(r,
- "doc_id",
- "key",
- "value");
- r = new DateStandardizationStream(r,
- "created_at",
- "ISO8601");
- r = new SnapshotReplayStream(vsd, r, "created_at", buffer_sz, fire_period);
+ // r = new MongoDBStream(connection_string, db_name, collection_name, vsd);
+ // r = new KeyValueToRelationalStream(r,
+ // "doc_id",
+ // "key",
+ // "value");
+ // r = new DateStandardizationStream(r,
+ // "created_at",
+ // "ISO8601");
+ // r = new SnapshotReplayStream(vsd, r, "created_at", buffer_sz, fire_period);
- mySetRealSource(r);
- }
- }
+ // mySetRealSource(r);
+ // }
+ // }
}
diff --git a/src/DataStreamSuite/DataStream/Primitives/Project/New/NewProjectStream.cs b/src/DataStreamSuite/DataStream/Primitives/Project/New/NewProjectStream.cs
index 299b05426..3b4e81539 100644
--- a/src/DataStreamSuite/DataStream/Primitives/Project/New/NewProjectStream.cs
+++ b/src/DataStreamSuite/DataStream/Primitives/Project/New/NewProjectStream.cs
@@ -6,6 +6,7 @@ namespace bts {
using System;
using System.Collections;
using System.Collections.Generic;
+ using System.Linq;
public class ProjectData {
// schema from the base stream
@@ -147,12 +148,23 @@ namespace bts {
mapped_to_orig_map.Add(mapped_col, orig_col);
orig_to_mapped.Add(orig_col, mapped_col);
} else {
- Log.Gen.error($"NewProjectStream unexpecetd token: {p_tok}");
+ if (input_schema.columnNames.Contains(p_tok)) {
+ string orig_col = p_tok;
+ string mapped_col = orig_col;
+ orig_to_mapped.Add(orig_col, mapped_col);
+ }
}
}
// make the new schema
List new_cols = new List();
foreach (ColumnInfo ci in input_schema.columnInfo) {
+ if (ci.colName == "id") {
+ // if there's a hidden col, other columns will show by default, but if cols are being
+ // projected epxlicitely, than others wil be hidden
+ ColumnInfo nc = new ColumnInfo(ci.colName, new_cols.Count, ci.colType);
+ new_cols.Add(nc);
+ continue;
+ }
if (removed_cols.Contains(ci.colName)) {
continue;
} else if (orig_to_mapped.ContainsKey(ci.colName)) {
@@ -160,8 +172,12 @@ namespace bts {
ColumnInfo nc = new ColumnInfo(mapped_name, new_cols.Count, ci.colType);
new_cols.Add(nc);
} else {
- ColumnInfo nc = new ColumnInfo(ci.colName, new_cols.Count, ci.colType);
- new_cols.Add(nc);
+ if (ci.colName == "id" || removed_cols.Count > 0) {
+ // if there's a hidden col, other columns will show by default, but if cols are being
+ // projected epxlicitely, than others wil be hidden
+ ColumnInfo nc = new ColumnInfo(ci.colName, new_cols.Count, ci.colType);
+ new_cols.Add(nc);
+ }
}
}
IBlueTableSchema new_schema = new BlueTableSchema(input_schema.tableName, new_cols.ToArray());
diff --git a/src/DataStreamSuite/DataStreamCore/BlueData/BlueDataUtility.cs b/src/DataStreamSuite/DataStreamCore/BlueData/BlueDataUtility.cs
index 5a1072853..f005df29d 100644
--- a/src/DataStreamSuite/DataStreamCore/BlueData/BlueDataUtility.cs
+++ b/src/DataStreamSuite/DataStreamCore/BlueData/BlueDataUtility.cs
@@ -1783,6 +1783,18 @@ namespace BCG {
#endregion
}
+ public static IBlueRow find_max(IEnumerable rows, params string[] cols_to_compare) {
+ if (rows == null || rows.Count() == 0)
+ return null;
+ IBlueRow max_row = rows.ElementAt(0);
+ foreach (IBlueRow r in rows) {
+ int c = compare_rows(r, max_row, cols_to_compare);
+ if (c > 0)
+ max_row = r;
+ }
+ return max_row;
+ }
+
public static int compare_rows(IBlueRow r1, IBlueRow r2, Dictionary sort_map) {
foreach (KeyValuePair p in sort_map) {
int comp = compare_send_dontcare_to_last(r1, p.Key, r2, p.Key, p.Value);
diff --git a/src/DataStreamSuite/DataStreamPublisher/DataStreamPublisherConfig.cs b/src/DataStreamSuite/DataStreamPublisher/DataStreamPublisherConfig.cs
index 2f4f52d45..5e6dba0e3 100644
--- a/src/DataStreamSuite/DataStreamPublisher/DataStreamPublisherConfig.cs
+++ b/src/DataStreamSuite/DataStreamPublisher/DataStreamPublisherConfig.cs
@@ -10,7 +10,8 @@ using System.IO;
using System.Text.Json;
using System.Text.Json.Serialization;
-public class DataStreamPublisherConfig {
+public class DataStreamPublisherConfig
+{
///
/// The login user for sparkd
///
@@ -38,6 +39,18 @@ public class DataStreamPublisherConfig {
[JsonPropertyName("publishers")]
public List? Publishers { get; set; }
+ [JsonPropertyName("mongodb_url")]
+ public string MONGO_DB_URL { get; set; }
+
+ [JsonPropertyName("mongodb_db_name")]
+ public string MONGO_DB_NAME { get; set; }
+
+ [JsonPropertyName("mongodb_db_collection")]
+ public string MONGO_DB_COLLECTION { get; set; }
+
+ [JsonPropertyName("benzinga_news_ttl")]
+ public string BENZINGA_NEWS_TTL { get; set; }
+
[JsonPropertyName("pubsub")]
public PubSubConfig? PubSub { get; set; }
@@ -49,11 +62,13 @@ public class DataStreamPublisherConfig {
public string? ConfigFile { get; set; }
public static DataStreamPublisherConfig Default() =>
- new() {
+ new()
+ {
PubSub = PubSubConfig.Default(),
};
- public void MergeWith(DataStreamPublisherConfig other) {
+ public void MergeWith(DataStreamPublisherConfig other)
+ {
User ??= other.User;
Password ??= other.Password;
Sparkd ??= other.Sparkd;
@@ -69,8 +84,10 @@ public class DataStreamPublisherConfig {
///
/// JsonSerializer.Deserialize returns null
///
- public void MergeWithConfigFile() {
- if (ConfigFile is null) {
+ public void MergeWithConfigFile()
+ {
+ if (ConfigFile is null)
+ {
return;
}
var cfgString = File.ReadAllText(ConfigFile);
diff --git a/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json b/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json
index a5e1aa6f7..67b031618 100644
--- a/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json
+++ b/src/DataStreamSuite/DataStreamQueryAnalyzer/Configurations/DataStreamQueryAnalyzer.json
@@ -1,61 +1,60 @@
-{
- "user": "bts",
- "pubsub": {
- "milliseconds_between_sub_info_updates": 1000,
- "milliseconds_between_time_slice_updates": 5000,
- "milliseconds_between_no_buffer_space_retries": 100,
- "retries_on_no_buffer_space": 1,
- "do_it_all_seconds_between_logs": 0,
- "heartbeat_interval": 15000,
- "missing_heartbeats_before_gc": 50,
- "use_managed_sockets": true,
- "state_update_period": 5000,
- "statistics_update_period": 5000,
- "disposed_connection_check_period": 1000,
- "message_processing_warning_threshold": 1000,
- "server_port": 12345,
- "publisher_routes": [
- {
- "name": "default",
- "server_address": "bluto://192.168.1.80:12345",
- "tag": ".*"
- }
- ],
- "subscriber_routes": [
- {
- "name": "default",
- "server_address": "bluto://192.168.1.80:12345",
- "tag": ".*"
- }
- ],
- "max_packet_size": 500000000,
- "blue_comm_must_die": true,
- "crushable_tags": [],
- "tag_type_to_assembly_map": [
- {
- "handler_class": "BCG.PubSubServerTagHandler",
- "assembly": "DataStreamCore.dll"
- }
- ],
- "server_queue": "PubSubQueue",
- "port_range": {
- "first": 30400,
- "last": 31700
- },
- "bind_to_random_port": false,
- "debug_tags": "^NOTAGS$",
- "latency_report_threshold_ms": 1500,
- "use_substations": false,
- "use_dynamic_clients": false,
- "is_substation": false,
- "internally_publish_sub_info": true,
- "use_managed_clients_in_native_code": false,
- "compressed_tags": "^NOTAGS$",
- "tags_to_report_compression_ratio": "^NOTAGS$",
- "disable_msmq": false,
- "use_prepublished_server_state": false,
- "ip_translation": null,
- "gc_activation_threshold": 0,
- "wait_period_after_gc": 1000
- }
+"user": "bts",
+"pubsub": {
+ "milliseconds_between_sub_info_updates": 1000,
+ "milliseconds_between_time_slice_updates": 5000,
+ "milliseconds_between_no_buffer_space_retries": 100,
+ "retries_on_no_buffer_space": 1,
+ "do_it_all_seconds_between_logs": 0,
+ "heartbeat_interval": 15000,
+ "missing_heartbeats_before_gc": 50,
+ "use_managed_sockets": true,
+ "state_update_period": 5000,
+ "statistics_update_period": 5000,
+ "disposed_connection_check_period": 1000,
+ "message_processing_warning_threshold": 1000,
+ "server_port": 12345,
+ "publisher_routes": [
+ {
+ "name": "default",
+ "server_address": "bluto://192.168.1.80:12345",
+ "tag": ".*"
+ }
+ ],
+ "subscriber_routes": [
+ {
+ "name": "default",
+ "server_address": "bluto://192.168.1.80:12345",
+ "tag": ".*"
+ }
+ ],
+ "max_packet_size": 500000000,
+ "blue_comm_must_die": true,
+ "crushable_tags": [],
+ "tag_type_to_assembly_map": [
+ {
+ "handler_class": "BCG.PubSubServerTagHandler",
+ "assembly": "DataStreamCore.dll"
+ }
+ ],
+ "server_queue": "PubSubQueue",
+ "port_range": {
+ "first": 30400,
+ "last": 31700
+ },
+ "bind_to_random_port": false,
+ "debug_tags": "^NOTAGS$",
+ "latency_report_threshold_ms": 1500,
+ "use_substations": false,
+ "use_dynamic_clients": false,
+ "is_substation": false,
+ "internally_publish_sub_info": true,
+ "use_managed_clients_in_native_code": false,
+ "compressed_tags": "^NOTAGS$",
+ "tags_to_report_compression_ratio": "^NOTAGS$",
+ "disable_msmq": false,
+ "use_prepublished_server_state": false,
+ "ip_translation": null,
+ "gc_activation_threshold": 0,
+ "wait_period_after_gc": 1000
+}
}
\ No newline at end of file
diff --git a/src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_APM.cs b/src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_APM.cs
index 044d63851..f7da41f20 100644
--- a/src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_APM.cs
+++ b/src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_APM.cs
@@ -36,14 +36,17 @@ using System.Linq;
using Bts.PubSub;
using Spark.Core;
-namespace bts {
- public class AsyncTCPConnection_APM : ITCPConnection, IWorker, IDisposable {
+namespace bts
+{
+ public class AsyncTCPConnection_APM : ITCPConnection, IWorker, IDisposable
+ {
// this class is used as the argument of the completion function for BeginSend/Receive
// serves as the base class for both Send and ReeceiveMessage classes.
// since both TCP sends and receives happen in multiple transfers, this
// provides a wrapper around copying the partial transfers into the complete buffer
- public class AsyncMessageBase {
+ public class AsyncMessageBase
+ {
readonly string msg_id = Guid.NewGuid().ToString();
// every BeginSend or BeginReceive has a unique key associated with it
public int key;
@@ -55,37 +58,45 @@ namespace bts {
public int transfer_ptr;
protected int _n_transfers = 0;
public bool completed_sync = false;
- public AsyncMessageBase(byte[] the_buffer, int the_offset) {
+ public AsyncMessageBase(byte[] the_buffer, int the_offset)
+ {
buffer = the_buffer;
transfer_ptr = the_offset;
}
- public int remaining_size {
- get {
+ public int remaining_size
+ {
+ get
+ {
return buffer.Length - transfer_ptr;
}
}
public int original_size => buffer.Length;
- public void advance_transfer_pointer(int advance_by) {
+ public void advance_transfer_pointer(int advance_by)
+ {
transfer_ptr += advance_by;
_n_transfers++;
- if (transfer_ptr > original_size) {
+ if (transfer_ptr > original_size)
+ {
bts.Log.Gen.error($"transfer went beyond initial size {ToString()}");
}
// TODO: log if it goes past the buffer
}
- public bool fully_transfered() {
+ public bool fully_transfered()
+ {
return transfer_ptr >= original_size;
}
- public override string ToString() {
+ public override string ToString()
+ {
return $"{GetType().ToString()} msg_id: {msg_id} key: {key} original_length: {original_size} transfer_ptr: {transfer_ptr} n_transfers: {_n_transfers} " +
$"fully_transfered: {fully_transfered()} completed_sync: {completed_sync}";
}
}
- public class AsyncSendMessage : AsyncMessageBase {
+ public class AsyncSendMessage : AsyncMessageBase
+ {
public string message_id;
public SendConfirmationDelegate on_send_del;
//private int _n_transfers = 0;
@@ -94,35 +105,43 @@ namespace bts {
: this(the_buffer, the_offset, String.Empty, null) { }
public AsyncSendMessage(byte[] the_buffer, int the_transfer_ptr, string the_message_id, SendConfirmationDelegate the_on_send_del)
- : base(the_buffer, the_transfer_ptr) {
+ : base(the_buffer, the_transfer_ptr)
+ {
message_id = the_message_id;
on_send_del = the_on_send_del;
}
- public override string ToString() {
+ public override string ToString()
+ {
return base.ToString();
}
}
- public class AsyncReceiveMessage : AsyncMessageBase {
- public enum ReadState {
+ public class AsyncReceiveMessage : AsyncMessageBase
+ {
+ public enum ReadState
+ {
reading_size,
reading_data
};
readonly ReadState read_state;
public AsyncReceiveMessage(int size, ReadState the_read_state) :
- this(new byte[size], 0, the_read_state) { }
+ this(new byte[size], 0, the_read_state)
+ { }
public AsyncReceiveMessage(byte[] the_buffer, int the_transfer_ptr, ReadState the_read_state)
- : base(the_buffer, the_transfer_ptr) {
+ : base(the_buffer, the_transfer_ptr)
+ {
read_state = the_read_state;
}
- public int read_int32(int start_idx = 0) {
+ public int read_int32(int start_idx = 0)
+ {
return BitConverter.ToInt32(buffer, start_idx);
}
- public byte[] read_bytes(int start_idx) {
+ public byte[] read_bytes(int start_idx)
+ {
byte[] rv = new byte[buffer.Length - start_idx];
Array.Copy(buffer, start_idx, rv, 0, rv.Length);
return rv;
@@ -131,16 +150,19 @@ namespace bts {
public bool is_reading_size() => read_state == ReadState.reading_size;
}
- public class ErrorMessage {
+ public class ErrorMessage
+ {
public readonly SocketError? socket_error = null;
public readonly Exception exception = null;
public readonly bool is_exception = false;
public readonly string context = String.Empty;
- public ErrorMessage(SocketError socket_error_, string context_) {
+ public ErrorMessage(SocketError socket_error_, string context_)
+ {
socket_error = socket_error_;
context = context_;
}
- public ErrorMessage(Exception exception_, string context_) {
+ public ErrorMessage(Exception exception_, string context_)
+ {
is_exception = true;
exception = exception_;
context = context_;
@@ -148,7 +170,8 @@ namespace bts {
public bool oom => exception != null && exception is OutOfMemoryException;
- public override string ToString() {
+ public override string ToString()
+ {
if (is_exception)
return $"ErrorMessage exception: {exception} context: {context}";
return $"socket_error: {socket_error} context: {context}";
@@ -174,13 +197,15 @@ namespace bts {
private bool _hb_received_yet = false, _data_received_yet = false, _hb_sent_once = false;
private DateTime _last_hb_received_time = new DateTime(1900, 1, 1);
- public AsyncTCPConnection_APM(Socket sock, TimeSpan hb_period, ITCPConnection.ConnectionSide connection_side) {
+ public AsyncTCPConnection_APM(Socket sock, TimeSpan hb_period, ITCPConnection.ConnectionSide connection_side)
+ {
_sock = sock;
_hb_period = hb_period;
_connection_side = connection_side;
_sock.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.DontRoute, true);
#region testing options
- switch (connection_side) {
+ switch (connection_side)
+ {
case ITCPConnection.ConnectionSide.Server:
TCPSocketUtils.set_server_socket_options(ref _sock);
break;
@@ -195,10 +220,13 @@ namespace bts {
_detect_connection_status_change();
}
- private void _detect_connection_status_change() {
- Action testing_func = () => {
+ private void _detect_connection_status_change()
+ {
+ Action testing_func = () =>
+ {
bool connected = _sock.Connected;
- while (true) {
+ while (true)
+ {
if (connected != _sock.Connected)
Log.Gen.error($"Connection status changed this: {ToString()} {connected} -> {_sock.Connected}");
connected = _sock.Connected;
@@ -208,26 +236,30 @@ namespace bts {
ThreadPool.QueueUserWorkItem(_ => testing_func());
}
- public override string ToString() {
+ public override string ToString()
+ {
return $"AsyncTCPConnection_APM side: {_connection_side} peer: {_peer_conn_info}, self: {_self_conn_info}";
}
#region ITCPConnection Members
- public void sendWithConfirmation(byte[] data, ref string conf_id, SendConfirmationDelegate del) {
+ public void sendWithConfirmation(byte[] data, ref string conf_id, SendConfirmationDelegate del)
+ {
conf_id = Guid.NewGuid().ToString();
byte[] msg = _build_data_message(data);
_begin_or_resume_sending(new AsyncSendMessage(msg, 0, conf_id, del));
}
- public void send(byte[] data) {
- Log.Gen.info($"AsyncTCPConnection_APM.send data: {AsyncTCPConnection_AsyncWrapper.to_hex(data, 0, data.Length)}");
+ public void send(byte[] data)
+ {
+ //Log.Gen.info($"AsyncTCPConnection_APM.send data: {AsyncTCPConnection_AsyncWrapper.to_hex(data, 0, data.Length)}");
byte[] msg = _build_data_message(data);
- Log.Gen.info($"AsyncTCPConnection_APM.send msg: {AsyncTCPConnection_AsyncWrapper.to_hex(msg, 0, msg.Length)}");
+ //Log.Gen.info($"AsyncTCPConnection_APM.send msg: {AsyncTCPConnection_AsyncWrapper.to_hex(msg, 0, msg.Length)}");
_begin_or_resume_sending(new AsyncSendMessage(msg, 0));
}
- public void register(ITCPMessageHandler messageHandler) {
+ public void register(ITCPMessageHandler messageHandler)
+ {
bts.Log.assert_d(_message_handler == null, "Message Handler is already assigned");
//bts.Log.Gen.info($"AsyncTCPConnection_APM for peer = {_peer_conn_info} being registered");
_message_handler = messageHandler;
@@ -236,15 +268,18 @@ namespace bts {
_read_next_msg();
}
- public RemoteConnectionInfo get_peer_address() {
+ public RemoteConnectionInfo get_peer_address()
+ {
return TCPSocketUtils.get_peer_address(_sock);
}
- public RemoteConnectionInfo get_self_address() {
+ public RemoteConnectionInfo get_self_address()
+ {
return TCPSocketUtils.get_peer_address(_sock);
}
- public bool isDisposed {
+ public bool isDisposed
+ {
get { return !_is_alive; }
}
@@ -253,12 +288,14 @@ namespace bts {
#region IWorker Members
bool _last_connected = false;
- public void doWork() {
+ public void doWork()
+ {
if (!_is_alive)
return;
return;
// disable hbs for now
- if (!_hb_sent_once) {
+ if (!_hb_sent_once)
+ {
bts.Log.Gen.info("Sending heartbeat to peer {_peer_conn_info} for the first time");
_hb_sent_once = true;
}
@@ -266,12 +303,14 @@ namespace bts {
_begin_or_resume_sending(new AsyncSendMessage(msg, 0));
}
- private static int _tid() {
+ private static int _tid()
+ {
var tid = System.Threading.Thread.CurrentThread.ManagedThreadId;
return tid;
}
- private static string _cb_txt(IAsyncResult res) {
+ private static string _cb_txt(IAsyncResult res)
+ {
return $"completed?: {res.IsCompleted} sync?: {res.CompletedSynchronously}";
}
@@ -279,10 +318,13 @@ namespace bts {
#region sending data
- private void _begin_or_resume_sending(AsyncSendMessage message) {
+ private void _begin_or_resume_sending(AsyncSendMessage message)
+ {
//bts.Log.Gen.info($"AsyncTCPConnection_APM._begin_or_resume_sending this: {ToString()} tid: {_tid()} message: {message} _is_alive: {_is_alive}");
- if (_is_alive) {
- try {
+ if (_is_alive)
+ {
+ try
+ {
SocketError begin_send_err;
// since we get message on on_send we dont need to keep the state around
message.key = _send_key++;
@@ -290,35 +332,45 @@ namespace bts {
out begin_send_err, new AsyncCallback(BeginSend_callback_function), message);
if (begin_send_err != SocketError.Success)
_handle_error(new ErrorMessage(begin_send_err, "_begin_or_resume_sending"));
- } catch (Exception ex) {
+ }
+ catch (Exception ex)
+ {
_handle_error(new ErrorMessage(ex, "_begin_or_resume_sending"));
}
}
}
- private void BeginSend_callback_function(IAsyncResult res) {
- if (res.CompletedSynchronously) {
+ private void BeginSend_callback_function(IAsyncResult res)
+ {
+ if (res.CompletedSynchronously)
+ {
ThreadPool.QueueUserWorkItem(_ => _on_send_async(res));
- } else
+ }
+ else
_on_send_async(res);
}
- private void _on_send_async(IAsyncResult res) {
+ private void _on_send_async(IAsyncResult res)
+ {
AsyncSendMessage message = res.AsyncState as AsyncSendMessage;
- if (_send_key_map.Contains(message.key)) {
+ if (_send_key_map.Contains(message.key))
+ {
bts.Log.Gen.warn($"AsyncTCPConnection_APM._on_send key: {message.key} already handled");
return;
}
_send_key_map.Add(message.key);
// bts.Log.Gen.info($"AsyncTCPConnection_APM._on_send: pre endSend this: {ToString()} tid: {_tid()} " +
// $"res: {_cb_txt(res)} message: {res.AsyncState} _is_alive: {_is_alive}");
- if (_is_alive) {
- try {
+ if (_is_alive)
+ {
+ try
+ {
SocketError send_err;
int bytes_sent = _sock.EndSend(res, out send_err);
//message.offset += bytesSent;
message.advance_transfer_pointer(bytes_sent);
- if (bytes_sent == 0) {
+ if (bytes_sent == 0)
+ {
bts.Log.Gen.error($"AsyncTCPConnection_APM._on_send this: {this} sent 0 bytes, throwing exception");
_handle_error(new ErrorMessage(new ApplicationException("Sent 0 bytes, peer socket is likely shutdown"),
"_on_send"));
@@ -326,22 +378,31 @@ namespace bts {
}
bts.Log.Gen.info($"AsyncTCPConnection_APM._on_send: post endSend this: {ToString()} tid: {_tid()} res: {_cb_txt(res)} " +
$"message: {res.AsyncState} bytes_sent: {bytes_sent} _is_alive: {_is_alive} send_err: {send_err}");
- if (send_err == SocketError.Success) {
+ if (send_err == SocketError.Success)
+ {
// successful send of some bytes
//if (message.offset == message.buffer.Length) {
- if (message.fully_transfered()) {
+ if (message.fully_transfered())
+ {
// fully sent, if confirmation needs to be sent out
- if (!String.IsNullOrEmpty(message.message_id)) {
+ if (!String.IsNullOrEmpty(message.message_id))
+ {
message.on_send_del(message.message_id, true);
}
- } else {
+ }
+ else
+ {
// there is more to send
_begin_or_resume_sending(message);
}
- } else {
+ }
+ else
+ {
_handle_error(new ErrorMessage(send_err, "_on_send"));
}
- } catch (Exception ex) {
+ }
+ catch (Exception ex)
+ {
_handle_error(new ErrorMessage(ex, "_on_send"));
}
}
@@ -352,8 +413,10 @@ namespace bts {
/// future references
///
///
- private byte[] _build_hb_message() {
- if (_cached_hb_message == null) {
+ private byte[] _build_hb_message()
+ {
+ if (_cached_hb_message == null)
+ {
BinaryWriter bw = new BinaryWriter(new MemoryStream());
// the total msg =[msg_sz, msg_type]
int msg_sz = 4;
@@ -365,7 +428,8 @@ namespace bts {
}
// writes size prefixed msg
- private byte[] _build_data_message(byte[] data) {
+ private byte[] _build_data_message(byte[] data)
+ {
// TODO, avoid unecessary memory creation (write directly into byte[] using BitConverter)
MemoryStream str = new MemoryStream();
BinaryWriter bw = new BinaryWriter(str);
@@ -384,28 +448,34 @@ namespace bts {
#region receiving data
- private void _read_next_msg() {
+ private void _read_next_msg()
+ {
_read_size();
}
- private void _read_size() {
+ private void _read_size()
+ {
_begin_or_resume_reading(new AsyncReceiveMessage(4, AsyncReceiveMessage.ReadState.reading_size));
}
- private void _read_data(int data_size) {
+ private void _read_data(int data_size)
+ {
// TODO, this could be a problem is the size is too big - maybe read it in chunks (later)
_begin_or_resume_reading(new AsyncReceiveMessage(data_size, AsyncReceiveMessage.ReadState.reading_data));
}
- private static string IAsyncResult_txt(IAsyncResult res) {
+ private static string IAsyncResult_txt(IAsyncResult res)
+ {
return $"IsCompleted: {res.IsCompleted} CompletedSynchronously: {res.CompletedSynchronously}";
}
- private void _begin_or_resume_reading(AsyncReceiveMessage rcv_msg) {
+ private void _begin_or_resume_reading(AsyncReceiveMessage rcv_msg)
+ {
//bts.Log.Gen.info($"AsyncTCPConnection_APM._begin_or_resume_reading: this: {ToString()} tid: {_tid()} message: {rcv_msg} _is_alive: {_is_alive}");
if (!_is_alive)
return;
- try {
+ try
+ {
SocketError read_sock_err;
rcv_msg.key = _receive_key++;
var res = _sock.BeginReceive(rcv_msg.buffer, rcv_msg.transfer_ptr,
@@ -413,50 +483,61 @@ namespace bts {
SocketFlags.None, out read_sock_err,
new AsyncCallback(BeginReceive_callback_function),
rcv_msg);
- if (read_sock_err != SocketError.Success) {
+ if (read_sock_err != SocketError.Success)
+ {
bts.Log.Gen.error($"_begin_or_resume_reading this: {this} error: {read_sock_err} rcv_msg: {rcv_msg}");
_handle_error(new ErrorMessage(read_sock_err, "_beging_or_resume_reading"));
}
- } catch (Exception ex) {
+ }
+ catch (Exception ex)
+ {
_handle_error(new ErrorMessage(ex, "_beging_or_resume_reading"));
}
}
- private void _log_incoming_msg(IAsyncResult res) {
- AsyncReceiveMessage rcv_message = res.AsyncState as AsyncReceiveMessage;
- var buff_txt = AsyncTCPConnection_AsyncWrapper.to_hex(rcv_message.buffer, 0, rcv_message.buffer.Length);
- Log.Gen.info($"BeginReceive_callback_function _log_incoming_msg buff: {buff_txt}");
+ private void _log_incoming_msg(IAsyncResult res)
+ {
+ // AsyncReceiveMessage rcv_message = res.AsyncState as AsyncReceiveMessage;
+ // var buff_txt = AsyncTCPConnection_AsyncWrapper.to_hex(rcv_message.buffer, 0, rcv_message.buffer.Length);
+ // Log.Gen.info($"BeginReceive_callback_function _log_incoming_msg buff: {buff_txt}");
}
// this is the function passed to BeginReceive, we
- private void BeginReceive_callback_function(IAsyncResult res) {
+ private void BeginReceive_callback_function(IAsyncResult res)
+ {
_log_incoming_msg(res);
- if (res.CompletedSynchronously) {
+ if (res.CompletedSynchronously)
+ {
ThreadPool.QueueUserWorkItem(_ => _on_read_async(res));
- } else
+ }
+ else
_on_read_async(res);
}
// this will always be called on the non main-thread. the reason this is on its own thread is that this function ends up
// calling _begin_or_resume_reading, which in tuns calls _on_read_async. So to not cause a stack overflow, this is broken
// always called on a different thread than _begin_or_resume_reading
- private void _on_read_async(IAsyncResult res) {
+ private void _on_read_async(IAsyncResult res)
+ {
AsyncReceiveMessage rcv_message = res.AsyncState as AsyncReceiveMessage;
- if (_read_key_map.Contains(rcv_message.key)) {
+ if (_read_key_map.Contains(rcv_message.key))
+ {
bts.Log.Gen.warn($"AsyncTCPConnection_APM._on_read key: {rcv_message.key} already handled");
return;
}
_read_key_map.Add(rcv_message.key);
if (!_is_alive)
return;
- try {
+ try
+ {
SocketError read_err;
Log.Gen.info($"AsyncTCPConnection_APM._on_read pre-EndReceive rcv_message: {rcv_message} res: {IAsyncResult_txt(res)}");
int bytes_read = _sock.EndReceive(res, out read_err);
rcv_message.advance_transfer_pointer(bytes_read);
Log.Gen.info($"AsyncTCPConnection_APM._on_read post-EndReceive rcv_message: {rcv_message} res: {IAsyncResult_txt(res)} bytes_read: {bytes_read}");
- if (bytes_read == 0) {
+ if (bytes_read == 0)
+ {
bts.Log.Gen.error($"AsyncTCPConnection_APM._on_read this: {this} read 0 bytes, throwing exception");
//bts.Log.Gen.error($"AsyncTCPConnection_APM._on_read this: {this} read 0 bytes, sleeping and ignoring");
//System.Threading.Thread.Sleep(50);
@@ -464,23 +545,32 @@ namespace bts {
return;
}
//return;
- if (read_err == SocketError.Success) {
- if (rcv_message.fully_transfered()) {
- if (rcv_message.is_reading_size()) {
+ if (read_err == SocketError.Success)
+ {
+ if (rcv_message.fully_transfered())
+ {
+ if (rcv_message.is_reading_size())
+ {
var msg_size = _on_read_size_completely(rcv_message);
- if (msg_size != null && msg_size.HasValue) {
+ if (msg_size != null && msg_size.HasValue)
+ {
_read_data(msg_size.Value);
}
return;
- } else {
+ }
+ else
+ {
// data is read
int msg_type = rcv_message.read_int32();
bts.Log.assert(msg_type == HB_MSG_SIGNAL || msg_type == DATA_MSG_SIGNAL);
- if (msg_type == DATA_MSG_SIGNAL) {
+ if (msg_type == DATA_MSG_SIGNAL)
+ {
_on_read_data_completely(rcv_message);
_read_next_msg();
return;
- } else if (msg_type == HB_MSG_SIGNAL) {
+ }
+ else if (msg_type == HB_MSG_SIGNAL)
+ {
_on_read_hb_completely(rcv_message);
_read_next_msg();
return;
@@ -488,45 +578,57 @@ namespace bts {
// go back to reading size for a new message
//_read_size();
}
- } else {
+ }
+ else
+ {
// msg has not been fully rd, we have already advanced the transfer ptr
_begin_or_resume_reading(rcv_msg: rcv_message);
}
- } else {
+ }
+ else
+ {
_handle_error(new ErrorMessage(read_err, "_on_read_async"));
}
- } catch (Exception ex) {
+ }
+ catch (Exception ex)
+ {
_handle_error(new ErrorMessage(ex, "_on_read_async"));
}
}
#endregion
- private int? _on_read_size_completely(AsyncReceiveMessage msg) {
+ private int? _on_read_size_completely(AsyncReceiveMessage msg)
+ {
bts.Log.assert(msg.is_reading_size());
bts.Log.assert(msg.fully_transfered());
bts.Log.assert(msg.original_size == 4);
var msg_sz = msg.read_int32();
- var buff_txt = AsyncTCPConnection_AsyncWrapper.to_hex(msg.buffer, 0, msg.buffer.Length);
- bts.Log.Gen.info($"AsyncTCPConnection_APM._on_read_size_completely buff: {buff_txt}");
+ //var buff_txt = AsyncTCPConnection_AsyncWrapper.to_hex(msg.buffer, 0, msg.buffer.Length);
+ //bts.Log.Gen.info($"AsyncTCPConnection_APM._on_read_size_completely buff: {buff_txt}");
bts.Log.Gen.info($"AsyncTCPConnection_APM._on_read_size_completely msg: {msg} msg_sz: {msg_sz}");
- if (msg_sz < 0 || msg_sz > _max_msg_size) {
+ if (msg_sz < 0 || msg_sz > _max_msg_size)
+ {
string err_msg = $"Encountered malformed message, prefixed_size: {msg_sz}";
_handle_error(new ErrorMessage(new ApplicationException(err_msg), "_on_read_size_completely"));
return null;
- } else {
+ }
+ else
+ {
// sz is determined, now we read that many bytes
//_read_data(msg_sz);
return msg_sz;
}
}
- private void _on_read_hb_completely(AsyncReceiveMessage msg) {
+ private void _on_read_hb_completely(AsyncReceiveMessage msg)
+ {
bts.Log.assert(!msg.is_reading_size());
bts.Log.assert(msg.fully_transfered());
bts.Log.assert(msg.original_size == 4);
bts.Log.assert(msg.read_int32(0) == HB_MSG_SIGNAL);
- if (!_hb_received_yet) {
+ if (!_hb_received_yet)
+ {
_hb_received_yet = true;
//bts.Log.Gen.info($"AsyncTCPConnection_APM HB received for the first time from peer {_peer_conn_info}");
}
@@ -534,7 +636,8 @@ namespace bts {
_message_handler.onData(new RemoteMessage(null, RemoteMessageType.Heartbeat, _peer_conn_info));
}
- private void _on_read_data_completely(AsyncReceiveMessage msg) {
+ private void _on_read_data_completely(AsyncReceiveMessage msg)
+ {
bts.Log.assert(!msg.is_reading_size());
bts.Log.assert(msg.fully_transfered());
var type = msg.read_int32(0);
@@ -543,7 +646,8 @@ namespace bts {
byte[] data = msg.read_bytes(4);
///callbacks can be made through thread pool to avoid this
///thread from being blocked, by this callback
- if (!_data_received_yet) {
+ if (!_data_received_yet)
+ {
_data_received_yet = true;
//bts.Log.Gen.info($"AsyncTCPConnection_APM data received for the first time from peer {_peer_conn_info}");
}
@@ -553,18 +657,24 @@ namespace bts {
#region IDisposable Members
- public void Dispose() {
+ public void Dispose()
+ {
// socket is closed and disposed here
- lock (this) {
+ lock (this)
+ {
bts.Log.Gen.info($"AsyncTCPConnection_APM.Dispose this: {this} Dispose called _dipose_impetus: {_dipose_impetus}");
- if (_is_alive) {
+ if (_is_alive)
+ {
_master_worker.removeWorker(this);
_is_alive = false;
bts.Log.Gen.info($"AsyncTCPConnection_APM.Dispose this: {this} being disposed, last HB was received at {_last_hb_received_time}");
- try {
+ try
+ {
bts.Log.Gen.info($"AsyncTCPConnection_APM.Dispose this: {this} shutting down socket now");
_sock.Shutdown(SocketShutdown.Both);
- } catch (Exception ex) {
+ }
+ catch (Exception ex)
+ {
String.Format("AsyncTCPConnection_APM.Dispose exception: {ex} thrown when closing socket");
if (ex is OutOfMemoryException)
throw ex;
@@ -577,8 +687,10 @@ namespace bts {
#endregion
private string _dipose_impetus = String.Empty;
- private void _handle_error(ErrorMessage err_msg) {
- lock (this) {
+ private void _handle_error(ErrorMessage err_msg)
+ {
+ lock (this)
+ {
Log.Gen.error($"AsyncTCPConnection_APM._handle_error this: {this} error_msg: {err_msg}");
if (err_msg.oom)
throw err_msg.exception;
diff --git a/src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_AsyncWrapper.cs b/src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_AsyncWrapper.cs
index bf6214ece..0a5104b82 100644
--- a/src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_AsyncWrapper.cs
+++ b/src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_AsyncWrapper.cs
@@ -1,4 +1,5 @@
-namespace bts {
+namespace bts
+{
using BCG;
using BCG.Utilities.General;
using System;
@@ -11,15 +12,19 @@ namespace bts {
// this ITCPConnection uses the TCPClient class instead of a raw socket (in the hope, that it would configure the socket correctly)
- public class AsyncTCPConnection_AsyncWrapper : ITCPConnection, IWorker, IDisposable {
- public class ConnectionID {
+ public class AsyncTCPConnection_AsyncWrapper : ITCPConnection, IWorker, IDisposable
+ {
+ public class ConnectionID
+ {
public readonly RemoteConnectionInfo self;
public readonly RemoteConnectionInfo peer;
public readonly Type conn_type;
- public override string ToString() {
+ public override string ToString()
+ {
return $"{conn_type} self: {self} peer: {peer}";
}
- public ConnectionID(RemoteConnectionInfo self_, RemoteConnectionInfo peer_, Type conn_type_) {
+ public ConnectionID(RemoteConnectionInfo self_, RemoteConnectionInfo peer_, Type conn_type_)
+ {
self = self_;
peer = peer_;
conn_type = conn_type_;
@@ -33,30 +38,34 @@ namespace bts {
private bool _is_alive = true;
ConnectionID _connection_id;
- public static string to_hex(byte[] byte_arr, int start_pos, int end_pos) {
- var len = end_pos - start_pos;
- StringBuilder hex_builder = new StringBuilder(len * 2);
- int i = 0;
+ // public static string to_hex(byte[] byte_arr, int start_pos, int end_pos) {
+ // var len = end_pos - start_pos;
+ // StringBuilder hex_builder = new StringBuilder(len * 2);
+ // int i = 0;
- foreach (byte b in byte_arr) {
- if (i >= start_pos && i < end_pos)
- hex_builder.Append($"{b:X2} ");
- i++;
- }
- return hex_builder.ToString().Trim();
- }
+ // foreach (byte b in byte_arr) {
+ // if (i >= start_pos && i < end_pos)
+ // hex_builder.Append($"{b:X2} ");
+ // i++;
+ // }
+ // return hex_builder.ToString().Trim();
+ // }
- public AsyncTCPConnection_AsyncWrapper(TcpClient tcp_client) {
+ public AsyncTCPConnection_AsyncWrapper(TcpClient tcp_client)
+ {
_tcp_client = tcp_client;
_nw_stream = _tcp_client.GetStream();
_connection_id = new ConnectionID(TCPSocketUtils.get_self_address(_tcp_client.Client), TCPSocketUtils.get_peer_address(_tcp_client.Client), GetType());
Log.Gen.info("AsyncTCPConnection_AsyncWrapper.cons connection: {_connection_id}");
}
- private void _detect_connection_status_change() {
- Action testing_func = () => {
+ private void _detect_connection_status_change()
+ {
+ Action testing_func = () =>
+ {
bool connected = _tcp_client.Connected;
- while (true) {
+ while (true)
+ {
if (connected != _tcp_client.Connected)
Log.Gen.error($"Connection status changed this: {ToString()} {connected} -> {_tcp_client.Connected}");
connected = _tcp_client.Connected;
@@ -69,15 +78,17 @@ namespace bts {
}
#region Sending
- public void _send_fixed_size_msg(byte[] buffer) {
- string buff_txt = to_hex(buffer, 0, buffer.Length);
- Log.Gen.info($"AsyncTCPConnection_AsyncWrapper._send_fixed_size_msg buffer: {buff_txt}");
+ public void _send_fixed_size_msg(byte[] buffer)
+ {
+ // string buff_txt = to_hex(buffer, 0, buffer.Length);
+ // Log.Gen.info($"AsyncTCPConnection_AsyncWrapper._send_fixed_size_msg buffer: {buff_txt}");
int total_bytes_sent = 0;
int n_bytes = buffer.Length;
int n_transfers = 0;
- while (_is_alive && total_bytes_sent < n_bytes) {
+ while (_is_alive && total_bytes_sent < n_bytes)
+ {
int remaining_bytes = n_bytes - total_bytes_sent;
_nw_stream.Write(buffer, total_bytes_sent, remaining_bytes);
total_bytes_sent += remaining_bytes;
@@ -86,8 +97,10 @@ namespace bts {
Log.Gen.info($"AsyncTCPConnection_AsyncWrapper._send_fixed_size_msg sent {buffer.Length} bytes in {n_transfers} transfers");
}
- private void _send_one_msg(byte[] msg) {
- try {
+ private void _send_one_msg(byte[] msg)
+ {
+ try
+ {
// the total msg = [msg_sz, msg_type, data]
int sz = msg.Length + 4;
MemoryStream str = new MemoryStream();
@@ -98,63 +111,77 @@ namespace bts {
str.Position = 0;
byte[] buff = str.ToArray();
_send_fixed_size_msg(buff);
- } catch (Exception ex) {
+ }
+ catch (Exception ex)
+ {
_handle_error(new ErrorMessage(ex, $"_send_one_msg call_stacK: {ex.StackTrace}"));
}
}
- private void _send_forever() {
- while (_is_alive) {
+ private void _send_forever()
+ {
+ while (_is_alive)
+ {
byte[] data = _send_queue.dequeue();
- if (data != null) {
+ if (data != null)
+ {
_send_one_msg(data);
}
}
}
private SyncQueue _send_queue = new SyncQueue();
- public void send(byte[] data) {
- var data_txt = to_hex(data, 0, data.Length);
- Log.Gen.info($"AsyncTCPConnection_AsyncWrapper.send data: {data_txt}");
- //Task.Run(() => _send_one_msg(data));
+ public void send(byte[] data)
+ {
+ // var data_txt = to_hex(data, 0, data.Length);
+ // Log.Gen.info($"AsyncTCPConnection_AsyncWrapper.send data: {data_txt}");
+ // //Task.Run(() => _send_one_msg(data));
//ThreadPool.QueueUserWorkItem(_ => _send_one_msg(data));
_send_queue.enqueue(data);
}
#endregion
- public void register(ITCPMessageHandler message_handler) {
+ public void register(ITCPMessageHandler message_handler)
+ {
_msg_handler = message_handler;
// starting thread pool threads to keep reading and sending
ThreadPool.QueueUserWorkItem(_ => _read_msgs_forever());
ThreadPool.QueueUserWorkItem(_ => _send_forever());
}
- public RemoteConnectionInfo get_peer_address() {
+ public RemoteConnectionInfo get_peer_address()
+ {
return _connection_id.peer;
}
- public RemoteConnectionInfo get_self_address() {
+ public RemoteConnectionInfo get_self_address()
+ {
return _connection_id.self;
}
public bool isDisposed { get { return !_is_alive; } }
- public void doWork() {
+ public void doWork()
+ {
// send hb
}
#region Reading
// This makes the blocking IO call, the callstack is _read_msgs_forever -> _read_one_message -> _read_fixed_size_msg
- private byte[] _read_fixed_size_msg(int n_bytes) {
- if (n_bytes <= 0) {
+ private byte[] _read_fixed_size_msg(int n_bytes)
+ {
+ if (n_bytes <= 0)
+ {
throw new ArgumentException("Number of bytes to read must be greater than zero.", nameof(n_bytes));
}
byte[] buffer = new byte[n_bytes];
int total_bytes_read = 0;
int n_transfers = 0;
- while (_is_alive && total_bytes_read < n_bytes) {
+ while (_is_alive && total_bytes_read < n_bytes)
+ {
int bytes_read = _nw_stream.Read(buffer, total_bytes_read, n_bytes - total_bytes_read);
- if (bytes_read <= 0) {
+ if (bytes_read <= 0)
+ {
throw new IOException("Read 0 bytes, the connection was closed before all data could be read.");
}
n_transfers++;
@@ -165,37 +192,44 @@ namespace bts {
return buffer;
}
- private class OneRecvMessage {
+ private class OneRecvMessage
+ {
public int msg_type;
public byte[] data;
- public OneRecvMessage(int msg_type_, byte[] data_) {
+ public OneRecvMessage(int msg_type_, byte[] data_)
+ {
msg_type = msg_type_;
data = data_;
}
- public OneRecvMessage(byte[] msg) {
+ public OneRecvMessage(byte[] msg)
+ {
MemoryStream str = new MemoryStream(msg);
BinaryReader br = new BinaryReader(str);
int msg_type = br.ReadInt32();
- if (msg_type == DATA_MSG_SIGNAL) {
+ if (msg_type == DATA_MSG_SIGNAL)
+ {
// TODO, polish it, too much memory bneing allocated
data = br.ReadBytes(msg.Length - 4);
}
}
- public override string ToString() {
+ public override string ToString()
+ {
string rv = $"OneMessage msg_type: {msg_type} data_len: {data.Length}";
return rv;
}
}
- private OneRecvMessage _read_one_message() {
+ private OneRecvMessage _read_one_message()
+ {
// read size first
byte[] sz_arr = _read_fixed_size_msg(4);
BinaryReader br = new BinaryReader(new MemoryStream(sz_arr));
br.BaseStream.Position = 0;
int msg_sz = br.ReadInt32();
Log.Gen.info($"AsyncTCPConnection_AsyncWrapper._read_one_message read msg_sz: {msg_sz}");
- if (msg_sz < 0) {
+ if (msg_sz < 0)
+ {
throw new IOException("Read 0 bytes, the connection was closed before all data could be read.");
}
byte[] msg = _read_fixed_size_msg(msg_sz);
@@ -205,19 +239,29 @@ namespace bts {
return rv;
}
- private void _read_msgs_forever() {
- while (_is_alive) {
- try {
+ private void _read_msgs_forever()
+ {
+ while (_is_alive)
+ {
+ try
+ {
OneRecvMessage one_msg = _read_one_message();
- if (one_msg.msg_type == DATA_MSG_SIGNAL) {
+ if (one_msg.msg_type == DATA_MSG_SIGNAL)
+ {
_msg_handler.onData(new RemoteMessage(one_msg.data, RemoteMessageType.Data, _connection_id.peer));
- } else if (one_msg.msg_type == HB_MSG_SIGNAL) {
+ }
+ else if (one_msg.msg_type == HB_MSG_SIGNAL)
+ {
// keep a last hb time an dudpate it here atomically
_msg_handler.onData(new RemoteMessage(null, RemoteMessageType.Heartbeat, _connection_id.peer));
- } else {
+ }
+ else
+ {
throw new ApplicationException($"Unknown message type: {one_msg.msg_type}");
}
- } catch (Exception ex) {
+ }
+ catch (Exception ex)
+ {
_handle_error(new ErrorMessage(ex, $"_read_msgs_forever, call_stack: {ex.StackTrace}"));
}
}
@@ -227,16 +271,19 @@ namespace bts {
public bool is_disposed => !_is_alive;
- public class ErrorMessage {
+ public class ErrorMessage
+ {
public readonly SocketError? socket_error = null;
public readonly Exception exception = null;
public readonly bool is_exception = false;
public readonly string context = String.Empty;
- public ErrorMessage(SocketError socket_error_, string context_) {
+ public ErrorMessage(SocketError socket_error_, string context_)
+ {
socket_error = socket_error_;
context = context_;
}
- public ErrorMessage(Exception exception_, string context_) {
+ public ErrorMessage(Exception exception_, string context_)
+ {
is_exception = true;
exception = exception_;
context = context_;
@@ -244,23 +291,30 @@ namespace bts {
public bool oom => exception != null && exception is OutOfMemoryException;
- public override string ToString() {
+ public override string ToString()
+ {
if (is_exception)
return $"ErrorMessage exception: {exception} context: {context}";
return $"socket_error: {socket_error} context: {context}";
}
}
- public void Dispose() {
+ public void Dispose()
+ {
// socket is closed and disposed here
- lock (this) {
- if (_is_alive) {
+ lock (this)
+ {
+ if (_is_alive)
+ {
//_master_worker.removeWorker(this);
_is_alive = false;
- try {
+ try
+ {
bts.Log.Gen.info($"AsyncTCPConnection_APM.Dispose this: {this} closing tcp_client nin an orderly fashion now");
_tcp_client.Close();
- } catch (Exception ex) {
+ }
+ catch (Exception ex)
+ {
String.Format("AsyncTCPConnection_APM.Dispose exception: {ex} thrown when closing socket");
if (ex is OutOfMemoryException)
throw ex;
@@ -270,8 +324,10 @@ namespace bts {
}
}
- private void _handle_error(ErrorMessage err_msg) {
- lock (this) {
+ private void _handle_error(ErrorMessage err_msg)
+ {
+ lock (this)
+ {
Log.Gen.error($"AsyncTCPConnection_APM._handle_error this: {this} error_msg: {err_msg}");
if (err_msg.oom)
throw err_msg.exception;
@@ -279,7 +335,8 @@ namespace bts {
}
}
- public override string ToString() {
+ public override string ToString()
+ {
return _connection_id.ToString();
}
}
diff --git a/src/DataStreamSuite/PubSubServer/Configurations/PubsubServer_Linux.json b/src/DataStreamSuite/PubSubServer/Configurations/PubsubServer_Linux.json
new file mode 100644
index 000000000..4c9a17a45
--- /dev/null
+++ b/src/DataStreamSuite/PubSubServer/Configurations/PubsubServer_Linux.json
@@ -0,0 +1,61 @@
+{
+ "user": "bts",
+ "pubsub": {
+ "milliseconds_between_sub_info_updates": 1000,
+ "milliseconds_between_time_slice_updates": 5000,
+ "milliseconds_between_no_buffer_space_retries": 100,
+ "retries_on_no_buffer_space": 1,
+ "do_it_all_seconds_between_logs": 0,
+ "heartbeat_interval": 15000,
+ "missing_heartbeats_before_gc": 50,
+ "use_managed_sockets": true,
+ "state_update_period": 5000,
+ "statistics_update_period": 5000,
+ "disposed_connection_check_period": 1000,
+ "message_processing_warning_threshold": 1000,
+ "server_port": 12345,
+ "publisher_routes": [
+ {
+ "name": "default",
+ "server_address": "bts-cloud://192.168.1.80:12345",
+ "tag": ".*"
+ }
+ ],
+ "subscriber_routes": [
+ {
+ "name": "default",
+ "server_address": "bts-cloud://192.168.1.80:12345",
+ "tag": ".*"
+ }
+ ],
+ "max_packet_size": 500000000,
+ "blue_comm_must_die": true,
+ "crushable_tags": [],
+ "tag_type_to_assembly_map": [
+ {
+ "handler_class": "BCG.PubSubServerTagHandler",
+ "assembly": "DataStreamCore.dll"
+ }
+ ],
+ "server_queue": "PubSubQueue",
+ "port_range": {
+ "first": 30400,
+ "last": 31700
+ },
+ "bind_to_random_port": false,
+ "debug_tags": "^NOTAGS$",
+ "latency_report_threshold_ms": 1500,
+ "use_substations": false,
+ "use_dynamic_clients": false,
+ "is_substation": false,
+ "internally_publish_sub_info": true,
+ "use_managed_clients_in_native_code": false,
+ "compressed_tags": "^NOTAGS$",
+ "tags_to_report_compression_ratio": "^NOTAGS$",
+ "disable_msmq": false,
+ "use_prepublished_server_state": false,
+ "ip_translation": null,
+ "gc_activation_threshold": 0,
+ "wait_period_after_gc": 1000
+ }
+}
\ No newline at end of file
diff --git a/src/DataStreamSuite/DataStream/Primitives/MongoDB/BenzingaNewStream.cs b/src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsStream.cs
similarity index 67%
rename from src/DataStreamSuite/DataStream/Primitives/MongoDB/BenzingaNewStream.cs
rename to src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsStream.cs
index 7660bcc66..d253fff9b 100644
--- a/src/DataStreamSuite/DataStream/Primitives/MongoDB/BenzingaNewStream.cs
+++ b/src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsStream.cs
@@ -7,26 +7,45 @@ namespace bts {
using System.Text.Json;
using System.Text.Json.Nodes;
[DataStreamUserRespresentation("benzingaNews")]
- public class BenzingaNewStream : ViewStream {
+ public class BenzingaNewsStream : ViewStream {
private IBlueTableSchema _schema = null;
private int _id = 0;
- // constructor takes in the raw mongo db stream, and the time to live (ttl), "7d" for example
- public BenzingaNewStream(ViewStream mongodb_raw_stream, string ttl) {
+
+
+ private static ViewStream _pre_process_raw_kvstream(ViewStream mongo_db_str) {
+ ViewStream r = new KeyValueToRelationalStream(mongo_db_str, "doc_id", "key", "value");
+ r = new DateStandardizationStream(r, "created_at", "ISO8601");
+ return r;
+ }
+
+ public BenzingaNewsStream(string connection_string, string db_name, string collection_name, string poll_period, string ttl)
+ : this(new MongoDBStream(connection_string, db_name, collection_name, poll_period), ttl) {
+ }
+
+ // public BenzingaNewsStream()
+ // : this(new MongoDBStream(DataStreamPub connection_string, db_name, collection_name, poll_period), ttl) {
+ // }
+
+ // assumes it gets a raw mongodb_raw_kv_stream with key value pairs
+ public BenzingaNewsStream(ViewStream mongodb_raw_kv_stream, string ttl) {
+ ViewStream output_stream = _pre_process_raw_kvstream(mongodb_raw_kv_stream);
// Hiding body (with project, columns can be hidden, or renamed)
- ViewStream output_stream = new ProjectStream(mongodb_raw_stream, "!body");
+ output_stream = new ProjectStream(output_stream, "!body");
// add column news_source = 'Benzinga', so we can append other news sources
output_stream = new AddColumnsStream(output_stream, @"news_source : Benzinga");
// now SuperMapStream is used to map one row into multiple, one per symbol,
// (SuperMapStream takes 2 lambdas, one to get the enw schema and another to
// return an array of rows, per one row)
- output_stream = new SuperMapStream(mongodb_raw_stream,
+ output_stream = new SuperMapStream(output_stream,
(IBlueTableSchema s) => {
// this lambda returns the new schema (ie with extra cols symbol, exchange, is_primary_exchange)
if (_schema == null) {
var out_schema = s;
out_schema = BlueDataUtility.remove_columns(out_schema, "securities");
- out_schema = BlueDataUtility.add_column(s, "doc_id_internal", typeof(string));
+ out_schema = BlueDataUtility.remove_columns(out_schema, "body");
+ out_schema = BlueDataUtility.remove_columns(out_schema, "tags");
+ out_schema = BlueDataUtility.add_column(out_schema, "doc_id_internal", typeof(string));
out_schema = BlueDataUtility.add_column(out_schema, "symbol", typeof(string));
out_schema = BlueDataUtility.add_column(out_schema, "exchange", typeof(string));
out_schema = BlueDataUtility.add_column(out_schema, "is_primary_exchange", typeof(bool));
diff --git a/src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsSymbolStateStream.cs b/src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsSymbolStateStream.cs
new file mode 100644
index 000000000..944c9b73a
--- /dev/null
+++ b/src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsSymbolStateStream.cs
@@ -0,0 +1,33 @@
+namespace bts {
+ using BCG;
+
+ using System.Collections.Generic;
+ using System.Linq;
+ using System.Security.Cryptography;
+
+ //[DataStreamUserRespresentation("benzingaNewsSymbols")]
+ public class BenzingaNewsSymbolStateStream : ViewStream, INonIncGroupByEvaluator {
+ public BenzingaNewsSymbolStateStream(ViewStream news_str) {
+ ViewStream r = new NonIncGroupByStream(news_str, this);
+ mySetRealSource(r);
+ }
+ public BenzingaNewsSymbolStateStream(string connection_string, string db_name, string collection_name, string poll_period, string ttl) :
+ this(new BenzingaNewsStream(connection_string, db_name, collection_name, poll_period, ttl)) {
+ }
+
+ public IBlueRowCollection evaluate(IDictionary> grouped_rows) {
+ BlueTable rv = null;
+ foreach (var kv in grouped_rows) {
+ var symbol_rows = kv.Value;
+ IBlueRow state_row = BlueDataUtility.find_max(symbol_rows, "created_at");
+ rv = rv == null ? new BlueTable(state_row.schema) : rv;
+ rv.addRow(state_row);
+ }
+ return rv;
+ }
+
+ public string[] get_group_by_cols() {
+ return new string[] { "symbol" };
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsSymbolsStream.cs b/src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsSymbolsStream.cs
new file mode 100644
index 000000000..b29ebbb6c
--- /dev/null
+++ b/src/DataStreamSuite/SharedStreams/Business/BenzingaNews/BenzingaNewsSymbolsStream.cs
@@ -0,0 +1,14 @@
+namespace bts {
+ using BCG;
+ [DataStreamUserRespresentation("benzingaNewsSymbols")]
+
+ public class BenzingaNewsSymbolsStream : ViewStream {
+ public BenzingaNewsSymbolsStream(ViewStream news_str) {
+ ViewStream rs = new DistinctRowStream(news_str, "symbol");
+ mySetRealSource(rs);
+ }
+ public BenzingaNewsSymbolsStream(string connection_string, string db_name, string collection_name, string poll_period, string ttl) :
+ this(new BenzingaNewsStream(connection_string, db_name, collection_name, poll_period, ttl)) {
+ }
+ }
+}
\ No newline at end of file
diff --git a/src/DataStreamSuite/SharedStreams/Config/SharedStreamConfig.cs b/src/DataStreamSuite/SharedStreams/Config/SharedStreamConfig.cs
new file mode 100644
index 000000000..adeb0316f
--- /dev/null
+++ b/src/DataStreamSuite/SharedStreams/Config/SharedStreamConfig.cs
@@ -0,0 +1,48 @@
+namespace Bts.DataStreamPublisher;
+
+using Bts.PubSub;
+
+using CommandLine;
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+public class SharedStreamConfig {
+
+ [JsonPropertyName("mongodb_url")]
+ public string MONGO_DB_URL { get; set; }
+
+ [JsonPropertyName("mongodb_db_name")]
+ public string MONGO_DB_NAME { get; set; }
+
+ [JsonPropertyName("mongodb_db_collection")]
+ public string MONGO_DB_COLLECTION { get; set; }
+
+ [JsonPropertyName("benzinga_news_ttl")]
+ public string BENZINGA_NEWS_TTL { get; set; }
+
+ [JsonPropertyName("pubsub")]
+ public PubSubConfig? PubSub { get; set; }
+
+
+ ///
+ /// Merges options with those defined in `ConfigFile` (does not overwrite
+ /// non-null existing values).
+ ///
+ ///
+ /// JsonSerializer.Deserialize returns null
+ ///
+ public void MergeWithConfigFile() {
+ // if (ConfigFile is null) {
+ // return;
+ // }
+ // var cfgString = File.ReadAllText(ConfigFile);
+ // var parsedCfg = JsonSerializer.Deserialize(cfgString) ??
+ // throw new InvalidOperationException($"Failed to parse json from {ConfigFile}");
+ // MergeWith(parsedCfg);
+ // ConfigFile = null;
+ }
+}