Merging changes

src/DataStreamSuite/DataStreamQueryAnalyzer/Program.cs

@@ -1,3 +1,79 @@
 using System;
+using BCG;
+using System.IO;
+using System.Runtime.Serialization;
+using System.Runtime.Serialization.Formatters.Binary;
+using bts;
+using mbts;
+using mbts.config;
+using bts.utils;
 
+// BTSNativeMethods.provide_command_line();
+// Configuration.initialize();
+
+bts.Log.Gen.info($"Starting DSQA");
 Console.WriteLine("Starting DataStreamQueryAnalyzer...");
+TestPubliser.publish();
+//TestPubliser.subscribe();
+
+class TestPubItem : IFormatter, IPubItem
+{
+   public DateTime send_time = DateTime.Now;
+   public object Deserialize(Stream str)
+   {
+      BinaryReader br = new BinaryReader(str);
+      var d = br.ReadDouble();
+      TestPubItem rv = new TestPubItem();
+      rv.send_time = DateTime.FromOADate(d);
+      return rv;
+   }
+   public void Serialize(Stream str, Object obj)
+   {
+      TestPubItem tpi = obj as TestPubItem;
+      BinaryWriter bw = new BinaryWriter(str);
+      bw.Write(tpi.send_time.ToOADate());
+   }
+   public System.Runtime.Serialization.SerializationBinder? Binder { get; set; }
+
+   public System.Runtime.Serialization.StreamingContext Context { get; set; }
+
+   public System.Runtime.Serialization.ISurrogateSelector? SurrogateSelector { get; set; }
+
+   public string getTag()
+   {
+      return "Test_Tag";
+   }
+}
+
+class TestPubliser
+{
+   public static void publish()
+   {
+      int n_pubs = 1000;
+      Publisher<TestPubItem> publisher = new Publisher<TestPubItem>();
+      for (int i = 0; i < n_pubs; i++)
+      {
+         TestPubItem pi = new TestPubItem();
+         pi.send_time = DateTime.Now;
+         bts.Log.Gen.info($"DSQA publishing tag: {pi.getTag()}");
+         publisher.publish(pi);
+         System.Threading.Thread.Sleep(100);
+         MasterWorker.getInstance(false).doWork();
+      }
+   }
+   public static void subscribe()
+   {
+      ManagedSubscriber ms = new ManagedSubscriber();
+      ms.subscribe("Test_Tag", new Handler(on_data));
+      while (true)
+      {
+         ms.doWork();
+      }
+   }
+   public static void on_data(IPubItem pi)
+   {
+      Console.WriteLine("Received data");
+   }
+}
+
+