Compare commits

..
Author SHA1 Message Date
John Colvin 554ceea2ee check pt 2024-07-07 22:19:25 +00:00
John Colvin d54b0437b0 Add config for PubSub tag handler factory 2024-07-05 13:50:39 +00:00
John Colvin 81f51352b4 devcontainer: Install dotnet8 expose ports for DataStream Web App 2024-07-04 17:33:37 +00:00
John Colvin 6f521f8f6a fix build: Copy libspark to net8.0 dir
For the DataStream Web App
2024-07-04 17:33:37 +00:00
John Colvin 7a0f36f686 fix build: Copy dll to all dirs in DOTNET_BINARY_DIR
Needed to wrap the var in double quotes
2024-07-04 17:33:37 +00:00
John Colvin a14aa58a24 build: Add scripts to publish data stream apps 2024-07-04 17:33:37 +00:00
John Colvin 667ec4ca10 Fix typo in log msg 2024-07-04 17:33:37 +00:00
John Colvin 7698fa15f6 Fix work scheduler issues 2024-07-04 13:32:22 -04:00
John Colvin e78b405139 refactor: Remove JsonPropertyName attributes from PubSubConfig
The ApplicationBuilder doesn't support these attributes when parsing config.
2024-07-04 13:32:22 -04:00
John Colvin bf96899fa3 refactor: Add config for IPubSubTagHandlerFactory 2024-07-04 13:32:22 -04:00
John Colvin e919aa9d05 fix: Make sure transient streams are attached to IServiceProvider
So they can retrieve services necessary to function (e.g., add themselves
to the master IWorkScheduler so doWork gets called, etc.)
2024-07-04 13:32:22 -04:00
John Colvin a05c7410dc refactor: Use DependencyInjection paradigm in DataStreamSuite
This allows us to create loosely coupled services with automatic
dependency resolution
2024-07-02 11:30:14 -04:00
John Colvin 6cc0ad4f66 Fix net7 -> net8 typo in publish_data_stream_web_app.py
The DataStream Web App is on net8
2024-07-01 23:36:04 -04:00
John Colvin 8fff414c4c Fix build scripts so they propagate the exit code back to the shell
I noticed this was broken because launch.json configurations would still start the debugger
even when the preLaunchTask (build) failed!
2024-07-01 23:36:04 -04:00
John Colvin 8421e4ae1a Change PubSub scheme prefix from 'bluto' -> 'bts-cloud' 2024-07-01 23:36:04 -04:00
Taha a8d0712e30 Added pubsub routes to the Title of DSQA 2024-07-01 23:36:04 -04:00
John Colvin aa36a7393c vscode: Add DataStreamPublisher debug cfg to skip build 2024-07-01 11:23:46 -04:00
John Colvin 7e0d5e4f59 build: Use separate cmd to build DataStreamSuite/QueryAnalyzer.sln
The DataStreamSuite.sln builds the new webapp now, which will
(eventually) replace DSQA. We'll keep it around for now, but with a
clean separation from the main build.
2024-07-01 11:23:46 -04:00
John Colvin d26e653a6b Tweak grid colors in DataStream web app 2024-07-01 11:23:46 -04:00
John Colvin 56e087ab8b refactor: Move DataStreamWebApp themes to their own js file 2024-07-01 11:23:45 -04:00
Taha c71bb8c301 Fixing the bug where messages were being sent by the pubsubserver out of order 2024-06-27 01:32:18 -04:00
John Colvin 44d6173ddf feat: Add DataStream WebApp
Included in DataStreamSuite.sln, but not DataStreamQueryAnalyzer.sln
2024-06-26 14:33:04 -04:00
John Colvin 322152c743 build: Copy libspark to all dotnet dirs in single cmake target
This is cleaner than the previous impl, which used separate build
targets to copy libspark to "net7.0", and "net7.0-windows" (dirs).

Note: I decided to clean this up now because I'm about to upgrade the
new DataStream WebApp to net8.0, and did not want to add yet another
build target to copy libspark to "net8.0".
2024-06-26 14:33:04 -04:00
John Colvin 9ab39587e2 fix: Parse ServerAddress as normal string
The "JsonValueOrArrayConverter<T>" doesn't work the way I thought it
would. When it is associated with a property via the JsonConverter
attribute, dotnet json parsing code throws an exception if "T" isn't
assignable to the property type.
2024-06-26 14:33:04 -04:00
John Colvin 72f59924ea fix build (api server): Change namespace to bts.websocket_core 2024-06-26 14:33:04 -04:00
John Colvin 98ab9741b5 fix build: Update vcpkg
On some vs versions, vcpkg cmake scripts would fail w/ an error saying
"Could not locate a complete Visual Studio instance"
2024-06-26 14:33:04 -04:00
John Colvin 11f0143845 Remove docker/windows again? 2024-06-26 14:33:04 -04:00
John Colvin 35d47f9cce Remove docker/linux again? 2024-06-26 14:33:04 -04:00
John Colvin 18662fcc0d Change externals url from ssh to https style
So we don't need ssh keys to clone it (the builtin credential manager
makes https authentication safe/easy)
2024-06-26 14:33:04 -04:00
John Colvin 16ffc48846 cleanup: Fix benign typos in build scripts
Change BUILD_SPARK_PY var name to BUILD_PY (because these scripts to actually build 'spark')
2024-06-26 14:33:04 -04:00
John Colvin 3cf388f041 fix build (warn): Only use Registry on windows 2024-06-26 14:33:03 -04:00
John Colvin 3e401a766d cleanup: Fix numerous build warnings in DataStreamSuite
Nearly all warnings were due to nullability issues, since the concept of
nullable reference types did not exist when the DataStream code was
originally written.

* PubSubClientMessage
  - Normalize indentation/whitespace
  - Fix some typos
  - Address nullability issues to fix build warnings
  - Modernize code

* Compression related files
  - ZipEntry
  - ZipFile
  - ZipInputStream
  - TarHeader

* RemoteMessage
  - make 'message' nullable to resolve warnings elsewhere (in calling code)

* PubSubSingleTagStateEvent.cs
* PubSubMessageReactor
* PubSubState
* SystemHeartbeatPublisher
* SystemStatsPublisher
* SingleTagData
* ManagedTCPMessageHub
* PubsubDeserializationController
* PublisherCore
* InflaterHuffmanTree
* AsyncTCPConnection_APM
* ThreadSafeQueue
* ManagedSubscriber
* SubscriberCore
* PubSubServerManagedTCPConnection
* AsyncTCPConnection_APM
* SuperMapStream
* MapStream
* TradeBarsStream
2024-06-26 14:33:03 -04:00
John Colvin b16ff50120 cleanup: Normalize indentation in TradeBarsStream 2024-06-26 14:33:03 -04:00
John Colvin 44564f7d56 cleanup: Disable nullability warnings
* BZip2InputStream
* TarInputStream
* Inflater
* InflaterDynHeader
* BZip2OutputStream
* DeflaterHuffman
* ParallelJoinTest
2024-06-26 14:33:03 -04:00
John Colvin eaa5580e89 cleanup: Modernize PubSubRoute code, json parsing 2024-06-26 14:33:03 -04:00
John Colvin 5de4e31880 cleanup: Remove unused DateTime grammar from exp parser 2024-06-26 14:33:03 -04:00
John Colvin 7cf90d95da Add JsonRpcRequestObservable, AsyncEnumerable 2024-06-26 14:33:03 -04:00
John Colvin ec36d84412 Add json facilities for DataStream websocket api 2024-06-26 14:33:03 -04:00
John Colvin 95aad773a2 refactor: Add handy WebsocketExtensions
* IsSocketStateOpen
* IsClosedProperly
2024-06-26 14:33:03 -04:00
John Colvin fb476de0a3 feat: Add JsonRpc types for websocket paradigm 2024-06-26 14:33:03 -04:00
John Colvin 41b42719ce cleanup: Remove empty file (clearly added by accident) 2024-06-26 14:33:03 -04:00
John Colvin c1f8aadd5d fix: Fix improper 'OfType' linq cast
Previously, animations were expressed through polymorphism (derived from the
base class CellAnimation), but the new impl uses composition. This 'OfType' call
was leftover from the polymorphism implementation, an attempt to linq cast
elements of base class (CellAnimation) to BackgroundAnimation.
2024-06-26 14:33:03 -04:00
John Colvin ad5e0c77fe editorconfig: Use standard c# capitalization rules
* Use 'PascalCase' for public members (instead of 'snake_case')
* Use 'camelCase' for private members (instead of '_snake_case')
2024-06-26 14:33:03 -04:00
John Colvin 90ee45db30 refactor: Add WebsocketClient class
Basically an ISubject (I/O) wrapper for IWebsocketConnectionGateway
2024-06-26 14:33:03 -04:00
John Colvin 6dd701dd33 refactor: Change WebsocketServerConfig.port to uint32
It should always be a number between 1-65535
2024-06-26 14:33:03 -04:00
John Colvin 456c953839 refactor: Rename WebsocketServerConfig proto pkg 'bts.websocket_core'
This matches the namespace used in spark_websocket csproj. Eventually,
I'd like to use conventional capitalization (TitleCase namespaces/types,
etc.), but let's shelf that until we have enough time/energy to do a
more comprehensive refactoring.
2024-06-26 14:33:03 -04:00
John Colvin 492f02eb21 refactor: Remove nullability constraint in GetValueOrDefault 2024-06-26 14:33:03 -04:00
John Colvin 41d2fd9c51 refactor: Use events to avoid circular deps for ws sync/gateway 2024-06-26 14:33:03 -04:00
John Colvin 59f3c446ba refactor: Let caller specify nullability of T in json parse fn 2024-06-26 14:33:03 -04:00
Taha 9aadda2429 some buf gixes + cleanup 2024-06-26 14:29:22 -04:00
Taha 57ac078469 PubSuMessageReactor empties the message queue (instead oif one elem ata a time) in the main loop 2024-06-26 14:28:52 -04:00
Taha c39fc70a64 PubSubServer's linux issues fixed 2024-06-26 14:28:51 -04:00
208 changed files with 3262 additions and 2780 deletions

No files matched your search

+8 -8
View File
@@ -11,10 +11,10 @@
"type": "coreclr",
"request": "launch",
"preLaunchTask": "Build DataStream WebApp (Debug)",
"program": "${workspaceFolder}/build/Debug/bin/net7.0/Bts.DataStreamWebApp.Server.dll",
"program": "${workspaceFolder}/build/Debug/bin/net8.0/Bts.DataStreamWebApp.Server.dll",
"args": [
"--config",
"${workspaceFolder}/src/DataStreamSuite/Bts.DataStreamWebApp.Server/Configurations/DevServer.json",
"--sparkd",
"http://localhost:8000/config",
],
"env": {
"ASPNETCORE_ENVIRONMENT": "Development",
@@ -25,11 +25,11 @@
"name": "Bts.DataStreamWebApp.Server (Skip Build)",
"type": "coreclr",
"request": "launch",
"program": "${workspaceFolder}/build/Debug/bin/net7.0/Bts.DataStreamWebApp.Server.dll",
"args": [
"--config",
"${workspaceFolder}/src/DataStreamSuite/Bts.DataStreamWebApp.Server/Configurations/DevServer.json",
],
"program": "${workspaceFolder}/build/Debug/bin/net8.0/Bts.DataStreamWebApp.Server.dll",
// "args": [
// "--config",
// "${workspaceFolder}/src/DataStreamSuite/Bts.DataStreamWebApp.Server/Configurations/DevServer.json",
// ],
"env": {
"ASPNETCORE_ENVIRONMENT": "Development",
"ASPNETCORE_HOSTINGSTARTUPASSEMBLIES": "Microsoft.AspNetCore.SpaProxy",
View File
Whitespace-only changes.
@@ -1,4 +1,5 @@
using System;
using System.Runtime.InteropServices;
using System.Runtime.Versioning;
using Microsoft.Win32;
@@ -15,11 +16,8 @@ namespace BCG
[SupportedOSPlatform("windows")]
public static double getValue (RegistryKey rk, string name)
{
object v = rk.GetValue (name);
if (v != null)
return double.Parse (v.ToString ());
else
return double.NaN;
var str = rk.GetValue(name)?.ToString();
return str is not null && double.TryParse(str, out var dbl) ? dbl : double.NaN;
}
}
}
@@ -4,7 +4,7 @@
font-weight: 400;
color-scheme: light dark;
color: rgba(255, 255, 255, 0.87);
color: rgba(0, 0, 0, 0.87);
background-color: #242424;
font-synthesis: none;
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<InvariantGlobalization>true</InvariantGlobalization>
@@ -12,8 +12,8 @@
<ItemGroup>
<PackageReference Include="CommandLineParser" Version="2.9.1" />
<PackageReference Include="Microsoft.AspNetCore.SpaProxy" Version="7.0.20" />
<PackageReference Include="Microsoft.AspNetCore.SpaServices.Extensions" Version="7.0.20" />
<PackageReference Include="Microsoft.AspNetCore.SpaProxy" Version="8.0.6" />
<PackageReference Include="Microsoft.AspNetCore.SpaServices.Extensions" Version="8.0.6" />
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
<PackageReference Include="Swashbuckle.AspNetCore" Version="6.4.0" />
</ItemGroup>
@@ -27,6 +27,7 @@
<ProjectReference Include="..\..\client\spark_api\spark_api.csproj" />
<ProjectReference Include="..\..\client\spark_core\spark_core.csproj" />
<ProjectReference Include="..\..\client\spark_websocket\spark_websocket.csproj" />
<ProjectReference Include="..\Bts.Extensions.DependencyInjection\Bts.Extensions.DependencyInjection.csproj" />
</ItemGroup>
<!-- <ItemGroup>
@@ -1,36 +1,15 @@
{
"Spark": {
"ConfigServer": "http://localhost:8000/config"
},
"SparkdCredentials": {
"User": "bts",
"Password": ""
},
"MasterWorker": {
"Interval": "00:00:00.1"
},
"ViewStreamDatabase": {
"Interval": "00:00:00.1"
},
"Sparkd": {
"Config": {
"ConfigServer": "http://localhost:8000/config"
},
"Credentials": {
"User": "bts",
"Password": ""
},
"EnableAdmin": true,
"EnableOrders": true,
"EnablePosition": true
},
"PubSub": {
"MillisecondsBetweenSubInfoUpdates": 1000,
"MillisecondsBetweenTimeSliceUpdates": 5000,
"MillisecondsBetweenNoBufferSpaceRetries": 100,
"RetriesOnNoBufferSpace": 1,
"DoItAllSecondsBetweenLogs": 0,
"HeartbeatInterval": 15000,
"MissingHeartbeatsBeforeGc": 50,
"UseManagedSockets": true,
"StateUpdatePeriod": 5000,
"StatisticsUpdatePeriod": 5000,
"DisposedConnectionCheckPeriod": 1000,
"MessageProcessingWarningThreshold": 1000,
"ServerPort": 12345,
"Routing": {
"PublisherRoutes": [
{
@@ -46,61 +25,9 @@
"TagRegExp": ".*"
}
]
},
"MaxPacketSize": 500000000,
"BlueCommMustDie": true,
"CrushableTags": [],
"TagTypeToAssemblyMap": {
"Entries": [
{
"HandlerClass": "BCG.PubSubServerTagHandler",
"Assembly": "DataStreamCore.dll"
}
]
},
"ServerQueue": "PubSubQueue",
"PortRange": {
"First": 30400,
"Last": 31700
},
"BindToRandomPort": false,
"DebugTags": "^NOTAGS$",
"LatencyReportThresholdMs": 1500,
"UseSubstations": false,
"UseDynamicClients": false,
"IsSubstation": false,
"InternallyPublishSubInfo": true,
"UseManagedClientsInNativeCode": false,
"CompressedTags": "^NOTAGS$",
"TagsToReportCompressionRatio": "^NOTAGS$",
"DisableMsmq": false,
"UsePrepublishedServerState": false,
"IpTranslation": null,
"GcActivationThreshold": 0,
"WaitPeriodAfterGc": 1000,
"TagHandlerFactory": {
"Handlers": [
{
"TagType": "DataStream",
"Assembly": "DataStreamCore",
"Class": "bts.PubsubServerStreamTagHandler"
},
{
"TagType": "System_Sub",
"Assembly": "PubSub",
"Class": "BCG.SystemSubscriptionMapPublisher"
},
{
"TagType": "System_Stats",
"Assembly": "PubSub",
"Class": "BCG.SystemStatsPublisher"
},
{
"TagType": ".*",
"Assembly": "DataStreamCore",
"Class": "bts.PubsubServerGenericDataHandler"
}
]
}
},
"ViewStreamDatabase": {
"Interval": "00:00:01"
}
}
@@ -1,30 +1,36 @@
namespace Bts.DataStreamWebApp.Server.Controllers;
using BCG;
using bts.api.core;
using Microsoft.AspNetCore.Mvc;
[Route("/ws")]
public class DataStreamSubscriptionController : ControllerBase {
readonly ILogger logger;
readonly PrimitiveManager primitiveManager;
readonly IServiceProvider serviceProvider;
public DataStreamSubscriptionController(ILogger<DataStreamSubscriptionController> logger) {
public DataStreamSubscriptionController(ILogger<DataStreamSubscriptionController> logger, PrimitiveManager primitiveManager, IServiceProvider serviceProvider) {
this.logger = logger;
this.primitiveManager = primitiveManager;
this.serviceProvider = serviceProvider;
logger.LogInformation("Creating {ControllerType}", GetType().Name);
}
[Route("/ws")]
public async Task Get(CancellationToken cancellationToken) {
public async Task Get() {
if (!HttpContext.WebSockets.IsWebSocketRequest) {
logger.LogError("Expected websocket request on path '{}'", HttpContext.Request.Path);
return;
}
logger.LogInformation("Establishing websocket connection...");
try {
using var webSocket = await HttpContext.WebSockets.AcceptWebSocketAsync();
logger.LogInformation("Websocket connection established.");
using var server = new DataStreamWebSocketServer(webSocket, logger);
await server.Run(cancellationToken);
} catch (Exception ex) {
logger.LogError("DataStreamSubscriptionController.Run threw exception: {Exception}", ex);
}
using var webSocket = await HttpContext.WebSockets.AcceptWebSocketAsync();
logger.LogInformation("Websocket connection established.");
var context = ServiceContext.create();
await context.login("bts", string.Empty);
var server = new DataStreamWebSocketServer(webSocket, logger, primitiveManager, serviceProvider);
await server.Run();
}
}
@@ -12,23 +12,22 @@ using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
class DataStreamSubscription : IDisposable, IHostedService {
class DataStreamSubscription : IDisposable {
public readonly ViewStream ViewStream;
public DataPacket? LastFired;
public readonly string Id;
readonly ILogger logger;
readonly WebSocket webSocket;
readonly DataPacketJsonSerializer dataPacketSerializer = new();
public DataStreamSubscription(string id, JsonRpcId requestId, ViewStream viewStream, WebSocket webSocket, ILogger logger) {
public DataStreamSubscription(string id, ViewStream viewStream, WebSocket webSocket, ILogger logger) {
ViewStream = viewStream;
RequestId = requestId;
Id = id;
this.webSocket = webSocket;
this.logger = logger;
}
public string Id { get; }
public JsonRpcId RequestId { get; }
public SubscriptionStatus Status { get; private set; }
public void Start() {
@@ -41,12 +40,6 @@ class DataStreamSubscription : IDisposable, IHostedService {
ViewStream.onChange += onChange;
}
public Task StartAsync(CancellationToken cancellationToken) {
cancellationToken.ThrowIfCancellationRequested();
Start();
return Task.CompletedTask;
}
public Task Clear() {
if (LastFired is null) {
return Task.CompletedTask;
@@ -65,12 +58,6 @@ class DataStreamSubscription : IDisposable, IHostedService {
ViewStream.onChange -= onChange;
}
public Task StopAsync(CancellationToken cancellationToken) {
cancellationToken.ThrowIfCancellationRequested();
Stop();
return Task.CompletedTask;
}
public void Dispose() =>
Stop();
@@ -1,5 +1,7 @@
namespace Bts.DataStreamWebApp.Server.Controllers;
using BCG;
using bts.websocket_core.json_rpc;
using Bts.DataStreamWebApp.Server.Messages;
@@ -9,22 +11,26 @@ using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
public class DataStreamWebSocketServer : IDisposable, IAsyncDisposable {
public class DataStreamWebSocketServer {
readonly ILogger logger;
readonly WebSocket webSocket;
readonly Dictionary<string, DataStreamSubscription> subscriptions = new();
readonly PrimitiveManager primitiveManager;
readonly IServiceProvider serviceProvider;
public DataStreamWebSocketServer(WebSocket webSocket, ILogger logger) {
public DataStreamWebSocketServer(WebSocket webSocket, ILogger logger, PrimitiveManager primitiveManager, IServiceProvider serviceProvider) {
this.webSocket = webSocket;
this.logger = logger;
this.primitiveManager = primitiveManager;
this.serviceProvider = serviceProvider;
}
public async Task<int> Run(CancellationToken cancellationToken) {
public async Task<int> Run() {
logger.LogInformation("DataStreamWebSocketServer running");
await foreach (var req in JsonRpcRequestAsyncEnumerable.Create(webSocket, logger, cancellationToken)) {
await foreach (var req in JsonRpcRequestAsyncEnumerable.Create(webSocket, logger)) {
await (req.Method switch {
"subscribe" => onSubscribe(req, cancellationToken),
"unsubscribe" => onUnsubscribe(req.Deserialize<UnsubscribeRequest>(), cancellationToken),
"subscribe" => onSubscribe(req),
"unsubscribe" => onUnsubscribe(req.Deserialize<UnsubscribeRequest>()),
_ => Task.CompletedTask,
});
}
@@ -32,51 +38,28 @@ public class DataStreamWebSocketServer : IDisposable, IAsyncDisposable {
return 0;
}
/// <summary>
/// Disposes internal subscriptions. To additionally notify the client of the subscription status
/// change(s), use <see cref="DisposeAsync"/>.
/// </summary>
public void Dispose() {
GC.SuppressFinalize(this);
disposeSubscriptions();
}
/// <summary>
/// Disposes internal subscriptions (like <see cref="Dispose"/>), then sends 'subscription status change'
/// notifications to the client (unlike <see cref="Dispose"/>).
/// </summary>
public async ValueTask DisposeAsync() {
GC.SuppressFinalize(this);
var finalSubs = subscriptions.Values.Select(s => (s.RequestId, s.Id)).ToList();
disposeSubscriptions();
// Send a 'subscription status change' msg to the client for each one.
foreach (var (requestId, id) in finalSubs) {
await sendSubscriptionStatus(requestId, id, "unsubscribe", SubscriptionStatus.Unsubscribed, CancellationToken.None);
}
}
async Task onSubscribe(JsonRpcRequest<JsonNode> req, CancellationToken cancellationToken) {
async Task onSubscribe(JsonRpcRequest<JsonNode> req) {
ArgumentNullException.ThrowIfNull(req.Id);
ArgumentException.ThrowIfNullOrEmpty(req.Method);
if (!SubscribeRequest.ParseQuery(req, out var vs, out var subId, out var errMsg, out var errCode)) {
if (!SubscribeRequest.ParseQuery(req, primitiveManager, out var vs, out var subId, out var errMsg, out var errCode)) {
logger.LogError("Invalid 'subscribe' request: {Reason}", errMsg);
var rpcError = new JsonRpcError<JsonNode>() {
await send(new JsonRpcError<JsonNode>() {
Code = (long)errCode,
Message = $"{errCode}: {errMsg}",
};
await send(rpcError, cancellationToken);
});
return;
}
if (subscriptions.TryGetValue(subId, out var oldSub)) {
oldSub.Dispose();
}
var sub = new DataStreamSubscription(subId, req.Id.Value, vs, webSocket, logger);
vs.Reset(serviceProvider);
var sub = new DataStreamSubscription(subId, vs, webSocket, logger);
subscriptions[subId] = sub;
await sendSubscriptionStatus(req.Id.Value, subId, req.Method, SubscriptionStatus.Subscribed, cancellationToken);
await sub.StartAsync(cancellationToken);
await sendSubscriptionStatus(req.Id.Value, subId, req.Method, SubscriptionStatus.Subscribed);
sub.Start();
}
async Task onUnsubscribe(JsonRpcRequest<UnsubscribeRequest> req, CancellationToken cancellationToken) {
async Task onUnsubscribe(JsonRpcRequest<UnsubscribeRequest> req) {
ArgumentNullException.ThrowIfNull(req.Id);
ArgumentException.ThrowIfNullOrEmpty(req.Method);
var subId = req.Params?.SubscriptionId ?? string.Empty;
@@ -87,35 +70,24 @@ public class DataStreamWebSocketServer : IDisposable, IAsyncDisposable {
logger.LogInformation("Disposing subscription");
sub.Dispose();
subscriptions.Remove(subId);
await sendSubscriptionStatus(req.Id.Value, subId, req.Method, SubscriptionStatus.Unsubscribed, cancellationToken);
await sendSubscriptionStatus(req.Id.Value, subId, req.Method, SubscriptionStatus.Unsubscribed);
}
async Task unsubscribe(JsonRpcId requestId, string subscriptionId, CancellationToken cancellationToken) {
await sendSubscriptionStatus(requestId, subscriptionId, "unsubscribe", SubscriptionStatus.Unsubscribed, cancellationToken);
}
Task sendSubscriptionStatus(JsonRpcId reqId, string subId, string method, SubscriptionStatus status, CancellationToken cancellationToken) {
Task sendSubscriptionStatus(JsonRpcId reqId, string subId, string method, SubscriptionStatus status) {
logger.LogInformation("Sending {Method} ack. SubscriptionId = {SubscriptionId}", method, subId);
var response = JsonRpcResponse.Create(reqId, new SubscriptionStatusMessage() {
return send(JsonRpcResponse.Create(reqId, new SubscriptionStatusMessage() {
SubscriptionId = subId,
Status = status,
});
return send(response, cancellationToken);
}));
}
Task send<T>(T message, CancellationToken cancellationToken) {
Task send<T>(T message) {
if (message is null) {
return Task.CompletedTask;
}
var str = JsonSerializer.Serialize(message);
var bytes = Encoding.UTF8.GetBytes(str);
logger.LogTrace("Sending data. Length = {Length}", bytes.Length);
return webSocket.SendAsync(bytes, WebSocketMessageType.Text, true, cancellationToken);
}
void disposeSubscriptions() {
logger.LogInformation("Disposing DataStreamWebSocketServer subscriptions. Count: {Count}", subscriptions.Count);
subscriptions.Values.DisposeAll();
subscriptions.Clear();
return webSocket.SendAsync(bytes, WebSocketMessageType.Text, true, CancellationToken.None);
}
}
@@ -18,7 +18,14 @@ public class SubscribeRequest {
[JsonPropertyName("subscription_id")]
public string? SubscriptionId { get; set; }
internal static bool ParseQuery(JsonRpcRequest<JsonNode> req, [NotNullWhen(true)] out ViewStream? stream, [NotNullWhen(true)] out string? subscriptionId, [NotNullWhen(false)] out string? errorMessage, out ErrorCode errorCode) {
internal static bool ParseQuery(
JsonRpcRequest<JsonNode> req,
PrimitiveManager primitiveManager,
[NotNullWhen(true)] out ViewStream? stream,
[NotNullWhen(true)] out string? subscriptionId,
[NotNullWhen(false)] out string? errorMessage,
out ErrorCode errorCode
) {
stream = null;
subscriptionId = null;
errorCode = ErrorCode.None;
@@ -50,7 +57,7 @@ public class SubscribeRequest {
}
string? exceptionMsg = null;
try {
stream = exp.getView(new Dictionary<string, StatementExpression>(0));
stream = exp.getView(new Dictionary<string, StatementExpression>(0), primitiveManager);
} catch (Exception ex) {
exceptionMsg = ex.ToString();
}
@@ -1,7 +1,9 @@
using Bts.DataStreamCore.Services;
using Bts.DataStreamWebApp.Server.Services;
using Bts.PubSub.Services;
using Spark.Api.Services;
using Spark.Core.Services;
using System.Reactive.Linq;
@@ -55,5 +57,8 @@ static void ConfigureServices(IServiceCollection services, IConfiguration config
services.AddMasterWorker(config.GetSection("MasterWorker"));
services.AddViewStreamDatabase(config.GetSection("ViewStreamDatabase"));
services.AddPubSubClient(config.GetSection("PubSub"));
services.AddSparkd(config.GetSection("Sparkd"));
services.AddPrimitiveManager();
services.AddSpark(config.GetSection("Spark"));
services.AddSparkd(config.GetSection("SparkdCredentials"));
services.AddHostedService<DataStreamWebAppService>();
}
@@ -0,0 +1,65 @@
namespace Bts.DataStreamWebApp.Server.Services;
using BCG;
using BCG.Utilities.General;
using bts;
using bts.api.core;
using Bts.DataStreamExpressions;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
/// <summary>
/// An IHostedService wrapper for ViewStream publishers. It provides a way for the Application to
/// start/stop the view publishers.
/// </summary>
class DataStreamWebAppService : IHostedService {
readonly DataStreamWebAppServerConfig config;
readonly ViewStreamDatabase vsd;
readonly TaskCompletionSource taskCompletionSource = new();
readonly CancellationTokenSource cancellationTokenSource = new();
public DataStreamWebAppService(
ViewStreamDatabase vsd,
IOptions<DataStreamWebAppServerConfig> config
) {
this.vsd = vsd;
this.config = config.Value;
}
public Task StartAsync(CancellationToken cancellationToken) {
// Task.Factory.StartNew(Run, cancellationToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
return Task.CompletedTask;
// void Run() {
// Log.Gen.info("DataStreamWebAppService STARTED");
// var defaultWaitTime = TimeSpan.FromMilliseconds(100);
// var innerCancellationToken = cancellationTokenSource.Token;
// var psConfig = config.PubSub ?? throw new InvalidOperationException("PubSub config is null");
// while (!innerCancellationToken.IsCancellationRequested) {
// vsd.doWork();
// masterWorker.doWork();
// Thread.Sleep(defaultWaitTime);
// }
// Log.Gen.info("DataStreamWebAppService STOPPED");
// taskCompletionSource.SetResult();
// }
}
public Task StopAsync(CancellationToken cancellationToken) {
Log.Gen.info("Stopping DataStreamWebAppService");
cancellationTokenSource.Cancel();
return Task.CompletedTask;
// return taskCompletionSource.Task;
}
}
@@ -0,0 +1,6 @@
{
"WebServer": {
"ContentRootPath": "build/Bts.DataStreamWebApp.Client/production",
"WebRootPath": "dist"
}
}
@@ -5,5 +5,36 @@
"Microsoft.AspNetCore": "Warning"
}
},
"AllowedHosts": "*"
"AllowedHosts": "*",
"Spark": {
"ConfigServer": "http://localhost:8000/config"
},
"SparkdCredentials": {
"User": "bts",
"Password": ""
},
"MasterWorker": {
"Interval": "00:00:00.1"
},
"PubSub": {
"Routing": {
"PublisherRoutes": [
{
"Name": "default",
"ServerAddress": "bts-cloud://127.0.0.1:12345",
"TagRegExp": ".*"
}
],
"SubscriberRoutes": [
{
"Name": "default",
"ServerAddress": "bts-cloud://127.0.0.1:12345",
"TagRegExp": ".*"
}
]
}
},
"ViewStreamDatabase": {
"Interval": "00:00:01"
}
}
@@ -0,0 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
</ItemGroup>
</Project>
@@ -0,0 +1,11 @@
namespace Bts.Extensions.DependencyInjection;
/// <summary>
/// This interface provides a way to move dependencies outside of the constructor (and accept them
/// via <see cref="IDependOn{T}.Reset(T?)"/> instead). It was designed to resolve circular
/// dependency issues among startup services.
/// </summary>
/// <typeparam name="T">The type of dependency</typeparam>
public interface IDependOn<T> {
void Reset(T? value);
}
@@ -0,0 +1,6 @@
#nullable enable
namespace Bts.Extensions.DependencyInjection;
public interface IDependOnKeyed<T> : IDependOn<T> {
string Key { get; }
}
@@ -0,0 +1,130 @@
namespace Bts.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using System;
using System.Linq;
public static class ServiceCollectionExtensions {
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="services"></param>
/// <returns></returns>
public static IServiceCollection EnsureSingleton<T>(this IServiceCollection services) where T : class =>
services.HasService<T>()
? services
: services.AddSingleton<T>();
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="services"></param>
/// <returns></returns>
public static IServiceCollection EnsureSingleton<T>(this IServiceCollection services, Func<IServiceProvider, T> implementationFactory) where T : class =>
services.HasService<T>()
? services
: services.AddSingleton(implementationFactory);
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T">The type of the service to add.</typeparam>
/// <typeparam name="TImplementation">The type of the implementation to use.</typeparam>
/// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
/// <returns>A reference to this instance after the operation has completed.</returns>
public static IServiceCollection EnsureSingleton<T, TImplementation>(this IServiceCollection services) where T : class where TImplementation : class, T =>
services.HasService<T>()
? services
: services.AddSingleton<T, TImplementation>();
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T">The type of the service to add.</typeparam>
/// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
/// <returns>A reference to this instance after the operation has completed.</returns>
public static IServiceCollection EnsureSingleton<T>(this IServiceCollection services, Func<IServiceProvider, object?, T> implementationFactory) where T : class =>
services.HasService<T>()
? services
: services.AddSingleton(implementationFactory);
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T">The type of the service to add.</typeparam>
/// <typeparam name="TImplementation">The type of the implementation to use.</typeparam>
/// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
/// <returns>A reference to this instance after the operation has completed.</returns>
public static IServiceCollection EnsureSingleton<T, TImplementation>(this IServiceCollection services, Func<IServiceProvider, TImplementation> implementationFactory) where T : class where TImplementation : class, T =>
services.HasService<T>()
? services
: services.AddSingleton(implementationFactory);
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T">The type of the service to add.</typeparam>
/// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
/// <returns>A reference to this instance after the operation has completed.</returns>
public static IServiceCollection EnsureKeyedSingleton<T>(this IServiceCollection services, object? key) where T : class =>
services.HasKeyedService<T>(key)
? services
: services.AddKeyedSingleton<T>(key);
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T">The type of the service to add.</typeparam>
/// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
/// <param name="implementationFactory">The factory that creates the service.</param>
/// <returns>A reference to this instance after the operation has completed.</returns>
public static IServiceCollection EnsureKeyedSingleton<T>(this IServiceCollection services, object? key, Func<IServiceProvider, object?, T> implementationFactory) where T : class =>
services.HasKeyedService<T>(key)
? services
: services.AddKeyedSingleton(key, implementationFactory);
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T">The type of the service to add.</typeparam>
/// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
/// <returns>A reference to this instance after the operation has completed.</returns>
public static IServiceCollection EnsureHostedService<T>(this IServiceCollection services) where T : class, IHostedService =>
services.HasService<T>()
? services
: services.AddHostedService<T>();
/// <summary>
/// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
/// already.
/// </summary>
/// <typeparam name="T">The type of the service to add.</typeparam>
/// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
/// <param name="implementationFactory">The factory that creates the service.</param>
/// <returns>A reference to this instance after the operation has completed.</returns>
public static IServiceCollection EnsureHostedService<T>(this IServiceCollection services, Func<IServiceProvider, T> implementationFactory) where T : class, IHostedService =>
services.HasService<T>()
? services
: services.AddHostedService(implementationFactory);
public static bool HasService<T>(this IServiceCollection services) =>
services.Any(s => s.ServiceType.IsAssignableTo(typeof(T)));
public static bool HasServiceImplementation<T, TImplementation>(this IServiceCollection services) =>
services.Any(s => s.ServiceType.IsAssignableTo(typeof(T)) && s.ImplementationType == typeof(TImplementation));
public static bool HasKeyedService<T>(this IServiceCollection services, object? key) =>
services.Any(s => Equals(s.ServiceKey, key) && s.ServiceType.IsAssignableTo(typeof(T)));
}
@@ -14,6 +14,7 @@
<ProjectReference Include="..\DataStreamCore\DataStreamCore.csproj" />
<ProjectReference Include="..\DataStreamExpressions\DataStreamExpressions.csproj" />
<ProjectReference Include="..\..\client\spark_core\spark_core.csproj" />
<ProjectReference Include="..\Bts.Extensions.DependencyInjection\Bts.Extensions.DependencyInjection.csproj" />
</ItemGroup>
<ItemGroup>
@@ -0,0 +1,35 @@
namespace Bts.DataStream;
using BCG;
using System;
using System.Collections.Generic;
public class BoundImportAttribute {
}
public class DataStreamActivator {
Dictionary<string, Type> myTypeLookupTable = new();
/// <summary>
/// Constructs and returns a ViewStream of type corresponding to the friendly name
/// </summary>
/// <param name="friendlyName">friendly primitive name </param>
/// <param name="args">args can be null or an empty ICollection if the paramterless constructor
/// is the desired one</param>
/// <returns></returns>
public ViewStream Activate(string friendlyName, IList<object>? args) {
object[]? constructorArgs = null;
if (args != null && args.Count > 0) {
constructorArgs = new object[args.Count];
args.CopyTo(constructorArgs, 0);
}
if (!myTypeLookupTable.TryGetValue(friendlyName, out var type)) {
throw new ApplicationException(friendlyName + " is not the friendly name of any of the loaded types");
}
return constructorArgs is null ?
Activator.CreateInstance(myTypeLookupTable[friendlyName] as Type) as ViewStream :
Activator.CreateInstance(myTypeLookupTable[friendlyName] as Type, constructorArgs) as ViewStream;
}
}
@@ -0,0 +1,7 @@
namespace Bts.DataStream;
using BCG;
public interface IViewStreamFactory {
public ViewStream? Create();
}
@@ -11,11 +11,8 @@ public class PandasScriptStream : ViewStream {
public PandasScriptStream(ViewStream vs, string file)
: this(vs, file, null) { }
public PandasScriptStream(ViewStream vs, string file, string? pythonExe)
: this(vs, file, pythonExe, ViewStreamDatabase.instance) { }
public PandasScriptStream(ViewStream vs, string file, string? pythonExe, ViewStreamDatabase vsd) {
mySetRealSource(new PandasStream(vs, readFile(), pythonExe, vsd));
public PandasScriptStream(ViewStream vs, string file, string? pythonExe) {
mySetRealSource(AttachStream<PandasStream>(new(vs, readFile(), pythonExe)));
string readFile() {
try {
@@ -1,7 +1,6 @@
namespace Bts.DataStream;
using BCG;
using BCG.Extensions;
using BCG.Utilities.General;
using bts;
@@ -23,7 +22,6 @@ public class PandasStream : ViewStream, IWorker {
readonly string inputFile = $"PandasStream_{Guid.NewGuid()}.csv";
readonly string outputFile = $"PandasStream_{Guid.NewGuid()}.csv";
readonly string pythonExe;
readonly ViewStreamDatabase vsd;
readonly CSVSaverStream inputCsvStream;
readonly Lazy<string> pythonCode;
IDictionary<int, IBlueRow>? lastRowMap = null;
@@ -34,12 +32,11 @@ public class PandasStream : ViewStream, IWorker {
IDisposable? dispose;
public PandasStream(ViewStream vs, string dataFrameExpression)
: this(vs, dataFrameExpression, null, ViewStreamDatabase.instance) { }
: this(vs, dataFrameExpression, null) { }
public PandasStream(ViewStream vs, string dataFrameExpression, string? pythonExe, ViewStreamDatabase vsd) {
public PandasStream(ViewStream vs, string dataFrameExpression, string? pythonExe) {
pythonExe ??= Environment.GetEnvironmentVariable("PYTHON_EXE") ?? @"C:\Users\johnc\GoogleDrive\Repos\bts\spark\.venv\Scripts\python.exe";
this.pythonExe = pythonExe;
this.vsd = vsd;
// This should fire an update when it writes to the output file
inputCsvStream = new CSVSaverStream(vs, inputFile, true, int.MaxValue, true);
pythonCode = new(getPythonCode);
@@ -59,7 +56,7 @@ import sys
}
protected override void myOnFirstSubscriber() {
vsd.addWorker(this);
base.myOnFirstSubscriber();
inputCsvStream.onChange += onInputChanged;
this.dispose = Disposable.Create(dispose);
@@ -68,7 +65,6 @@ import sys
void dispose() {
inputCsvStream.onChange -= onInputChanged;
vsd.removeWorker(this);
deleteFile(inputFile);
deleteFile(outputFile);
}
@@ -85,8 +81,10 @@ import sys
}
}
protected override void myOnLastSubscriber() =>
protected override void myOnLastSubscriber() {
base.myOnLastSubscriber();
Disposable.dispose(ref dispose);
}
static void executePythonCode(string pythonCode, string? pythonExe = null) {
pythonExe ??= "python";
@@ -1,16 +1,19 @@
using BCG.Utilities.General;
using Bts.PubSub;
using Microsoft.Extensions.DependencyInjection;
using System;
namespace BCG {
public class TCPConnectionTestStream : ViewStream, IWorker, ITCPConnectionHandler, ITCPMessageHandler {
public class TCPConnectionTestStream : ViewStream, IParameterizedWorker, ITCPConnectionHandler, ITCPMessageHandler {
class State {
public int n_sends = 0;
public int n_recvs = 0;
public int n_sends = 0;
public int n_recvs = 0;
}
private readonly string _ip;
private readonly int _port;
private readonly ViewStreamDatabase _vsd;
State _state = new State();
private readonly int _msg_size;
@@ -18,14 +21,12 @@ namespace BCG {
private TimeSpan _send_period;
private ITCPConnection _connection = null;
private byte[] _buff;
public TCPConnectionTestStream(string ip, int port, int msg_size, int send_period_seconds) :
this(ip, port, msg_size, send_period_seconds, ViewStreamDatabase.instance) { }
public TCPConnectionTestStream(string ip, int port, int msg_size, int send_period_seconds, ViewStreamDatabase vsd) {
WorkActivator workActivator;
public TCPConnectionTestStream(string ip, int port, int msg_size, int send_period_seconds) {
_ip = ip;
_port = port;
_msg_size = msg_size;
_vsd = vsd;
_schema = BlueDataUtility.form_schema("TCPClient",
Tuple.Create("id", TypeCode.Int32),
Tuple.Create("n_sends", TypeCode.UInt32),
@@ -35,16 +36,21 @@ namespace BCG {
for (int i = 0; i < _buff.Length; i++) {
_buff[i] = (byte)i;
}
workActivator = new(this);
}
TimeSpan IParameterizedWorker.Interval => _send_period;
protected override void OnServiceProvider(IServiceProvider serviceProvider) {
workActivator.OnWorkScheduler(serviceProvider.GetRequiredService<IWorkScheduler>());
}
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, _send_period);
var server_rci = new RemoteConnectionInfo(_ip, _port);
bts.AsyncTCPConnector_APM.connect_to(server_rci, this);
workActivator.Start();
}
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
workActivator.Stop();
}
public void doWork() {
@@ -60,20 +66,20 @@ namespace BCG {
}
public void onConnection(ITCPConnection connection) {
_connection = connection;
_connection.register(this);
_connection = connection;
_connection.register(this);
}
public bool isDisposed { get; }
public void onData(RemoteMessage msg) {
bts.Log.Gen.info($"TCPConnectionTestStream received msg: {msg}");
_state.n_recvs++;
bts.Log.Gen.info($"TCPConnectionTestStream received msg: {msg}");
_state.n_recvs++;
}
public void Dispose() {
workActivator.Stop();
}
}
}
}
@@ -2,15 +2,14 @@ using System;
using System.Collections.Generic;
using Bts.DataStreamExpressions;
using BCG.Utilities.General;
using bts;
using Bts.PubSub;
using Bts.DataStreamCore;
using Bts.PubSub;
namespace BCG {
// listens to the requests posted on "cloud_streams" request table, and
// if unfulfilled starts a ViewPubsliher
public class RemoteQueryManager : IWorker {
public class RemoteQueryManager : IParameterizedWorker {
public class StreamPubUnit {
public string name;
public string stream_expression;
@@ -26,12 +25,13 @@ namespace BCG {
view_publisher = view_publisher_;
}
}
static ViewStream? build_stream(string stream_expression)
static ViewStream? build_stream(string stream_expression, PrimitiveManager primitiveManager, IServiceProvider serviceProvider)
{
try
{
var exp = DataStreamExpressionsInterpreter.Evaluate(stream_expression);
var rv = exp.getView(new Dictionary<string, StatementExpression>(0));
var rv = exp.getView(new Dictionary<string, StatementExpression>(0), primitiveManager);
rv?.Reset(serviceProvider);
return rv;
} catch (Exception ex) {
@@ -43,18 +43,24 @@ namespace BCG {
_stream_pub_map = new Dictionary<string, StreamPubUnit>();
private ViewStream _stream = new TableStream("cloud_streams");
readonly IPubTableBroker _broker;
readonly IManagedSubscriber _sub;
private readonly ManagedSubscriber _sub;
readonly PrimitiveManager primitiveManager;
readonly IServiceProvider serviceProvider;
public RemoteQueryManager(IManagedSubscriber managedSubscriber, IPubTableBroker pubTableBroker) {
_sub = managedSubscriber;
_broker = pubTableBroker;
public RemoteQueryManager(ManagedSubscriber sub, IPubTableBroker broker, PrimitiveManager primitiveManager, IServiceProvider serviceProvider) {
_sub = sub;
_broker = broker;
this.primitiveManager = primitiveManager;
this.serviceProvider = serviceProvider;
_stream.onChange += _on_str;
_sub.Subscribe("RemoteRequest", (string tag, byte[] data) => {
_sub.subscribe("RemoteRequest", (string tag, byte[] data) => {
RemoteRequest req = RemoteRequest.construct_from_bytes(data);
_spawn_remote_stream(req.published_table_name, req.stream_expression, false);
});
}
TimeSpan IParameterizedWorker.Interval => TimeSpan.FromSeconds(1);
private void _on_str(object obj)
{
if (obj is not DataPacket dp) {
@@ -73,7 +79,11 @@ namespace BCG {
private void _spawn_remote_stream(string name, string stream_expression, bool persist) {
bts.Log.Gen.info($"CloudStreamManager.assign name: {name} stream_expression: {stream_expression}");
ViewStream stream = build_stream(stream_expression);
var stream = build_stream(stream_expression, primitiveManager, serviceProvider);
if (stream is null) {
Log.Gen.error($"Failed to spawn remote stream. stream_expression: {stream_expression}");
return;
}
_expose_to_ws(name, stream);
if (persist) {
stream = new PersistorStream(stream, name, "C:\\Temp\\", 1000000); // 10MBs
@@ -1,7 +1,7 @@
using BCG;
using bts;
using Microsoft.Extensions.DependencyInjection;
using System;
namespace BCG {
@@ -9,7 +9,7 @@ namespace BCG {
[DataStreamUserRespresentation("remoteQuery")]
public class RemoteQueryStream : ViewStream {
private readonly string _name, _stream_exp;
private PublisherCore _publisher_core = new PublisherCore();
IPublisherCore? _publisher_core_backing;
private ViewStream _source_str;
public RemoteQueryStream(params string[] lines) {
@@ -20,7 +20,14 @@ namespace BCG {
}
IPublisherCore? _publisher_core =>
_publisher_core_backing ??= ServiceProvider?.GetService<IPublisherCore>();
protected override void myOnFirstSubscriber() {
if (_publisher_core is null) {
throw new InvalidOperationException(
"Unable to initialize PublisherStream because PublisherCore is null");
}
RemoteRequest rem_req = new RemoteRequest(_stream_exp, _name);
byte[] rem_req_bytes = rem_req.serialize();
_publisher_core.publish("System", "RemoteRequest", rem_req_bytes);
@@ -15,7 +15,7 @@ namespace BCG {
public AddColumnsStream(ViewStream vs, string col_def) {
_new_col = parse_exp(col_def);
ViewStream rs = new MapStream(vs, (IBlueTableSchema s) => _get_schema(s), (IBlueRow r) => _map(r));
ViewStream rs = AttachStream(new MapStream(vs, _get_schema, _map));
mySetRealSource(rs);
}
@@ -91,10 +91,14 @@ namespace BCG
}
_increment_values = increment_values_del;
_get_column_prefix = get_column_prefix_del;
if(only_fire_once)
mySetRealSource(new NonIncGroupByStream(new OnlyFireOnceStream(stream), this));
else
mySetRealSource(new NonIncGroupByStream(stream, this));
if(only_fire_once) {
ViewStream vs = AttachStream(new OnlyFireOnceStream(stream));
vs = AttachStream(new NonIncGroupByStream(vs, this));
mySetRealSource(vs);
} else {
ViewStream vs = AttachStream(new NonIncGroupByStream(stream, this));
mySetRealSource(vs);
}
}
private void _set_schema(List<string> tc_prefixes, Pair<string, Type>[] group_by_col_info, string table_name)
@@ -241,4 +245,4 @@ namespace BCG
return values;
}
}
}
}
@@ -14,7 +14,7 @@ namespace BCG
public CaseChangeStream(ViewStream input, bool toUpper, params string[] columnsToChangeCase)
{
myCaseChangingColumns = columnsToChangeCase;
mySetRealSource(new SuperMapStream(input, this));
mySetRealSource(AttachStream(new SuperMapStream(input, this)));
myIsUpperCase = toUpper;
}
@@ -32,10 +32,10 @@ namespace BCG
{
mySetRealSource (
new MapStream (sourceStream,
AttachStream(new MapStream (sourceStream,
isConcatenatedDataConfined ? new ConcatenateColumnsMapperForSameDataSet (columnDefs) :
new ConcatenateColumnsMapper (columnDefs)
)
))
);
}
}
@@ -10,7 +10,7 @@ namespace BCG
public AverageCollectorStream(ViewStream stream, params string[] columnsToAverage)
{
myAveragingColumns = columnsToAverage;
mySetRealSource(new CrushingCollectorStream(stream, this));
mySetRealSource(AttachStream(new CrushingCollectorStream(stream, this)));
}
private string[] myAveragingColumns;
@@ -16,8 +16,8 @@ namespace BCG
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));
mySetRealSource (AttachStream(new GroupByStream (stream,
new DistinctRowEvaluator (groupByCols.ToArray (typeof(string)) as string[]), false)));
}
}
}
@@ -3,25 +3,20 @@ using System.Collections.Generic;
using System.Linq;
using System.Text;
using BCG.Utilities.General;
namespace BCG
{
public class DateBasedFilter: IDynamicStreamFilter, IWorker
{
private ViewStreamDatabase myVSD;
private string myDateColumn;
private int myNDays;
private DateTime myReference;
private DateTime myTodaysDate;
public DateBasedFilter(string dateColumn, int nDays)
: this(dateColumn, nDays, ViewStreamDatabase.instance) { }
public DateBasedFilter(string dateColumn, int nDays, ViewStreamDatabase vsd)
{
myDateColumn = dateColumn;
myNDays = nDays;
myVSD = vsd;
myVSD.addWorker(this);
myTodaysDate = DateTime.Today;
myReference = myTodaysDate.AddDays(-1 * myNDays);
}
@@ -1,7 +1,12 @@
using BCG.Utilities.General;
using Bts.DataStream;
using Bts.PubSub;
using mbts.config;
using Spark.Core.Disposables;
using System;
using System.Text.RegularExpressions;
@@ -141,6 +146,7 @@ namespace BCG
private BlueTable myLastFiredRows;
private IBlueRowCollection myLastReceivedRows;
private IStreamFilter myFilter;
private IDisposable? filterWorkerSubscription;
private ViewStream myParentStream;
private string myFilterString;
private BCG.StreamWorld.Addition.FilterStream myFilterStream;
@@ -168,7 +174,7 @@ namespace BCG
if (myUseNewFilter)
{
myFilterStream = new BCG.StreamWorld.Addition.FilterStream(vs, filter);
mySetRealSource(myFilterStream);
mySetRealSource(AttachStream(myFilterStream));
}
else
{
@@ -260,6 +266,7 @@ namespace BCG
}
else
{
Disposable.dispose(ref filterWorkerSubscription);
myFilter = value;
if (numberOfSubscribers > 0)
{
@@ -274,6 +281,9 @@ namespace BCG
BlueDataUtility.computeDifferences(nRows, myLastFiredRows, out dRows, out aRows);
myLastFiredRows = nRows;
myFireEvent(new DataPacket(nRows, aRows, dRows));
if (myFilter is IWorker filterWorker) {
filterWorkerSubscription = ViewStreamDatabase.Subscribe(filterWorker);
}
}
}
}
@@ -294,6 +304,14 @@ namespace BCG
return GetType().ToString() + " " + myFilterString;
}
protected override void OnViewStreamDatabase(IWorkScheduler viewStreamDatabase) {
base.OnViewStreamDatabase(viewStreamDatabase);
var w = this as IWorker;
if (w is not null) {
filterWorkerSubscription = viewStreamDatabase.Subscribe(w);
}
}
private void myHandleFilterChange()
{
filter = myFilter;
@@ -4,11 +4,13 @@ using System.Linq;
using System.Text;
using BCG.Utilities.General;
using Bts.PubSub;
namespace BCG
{
public class SlidingWindowFilter: IDynamicStreamFilter, IWorker
{
private ViewStreamDatabase myVSD;
private IWorkScheduler workScheduler;
private string myDateColumn;
private int myNSecs;
private DateTime myReference;
@@ -17,14 +19,9 @@ namespace BCG
private DateTime myLastTimeFilteredApplied;
public SlidingWindowFilter(string dateColumn, int nSecs)
: this(dateColumn, nSecs, ViewStreamDatabase.instance) { }
public SlidingWindowFilter(string dateColumn, int nSecs, ViewStreamDatabase vsd)
{
myDateColumn = dateColumn;
myNSecs = nSecs;
myVSD = vsd;
myVSD.addWorker(this);
myCurrentTime = DateTime.Now;
myReference = myCurrentTime.AddSeconds(-1 * myNSecs);
}
@@ -55,5 +52,9 @@ namespace BCG
}
}
public void Dispose() {
workScheduler.Remove(this);
}
}
}
@@ -4,7 +4,7 @@ using System.Text.RegularExpressions;
using BCG.Utilities.General;
namespace BCG {
public class TimeBasedFilter : IDynamicStreamFilter, IWorker, IDisposable {
public class TimeBasedFilter : IDynamicStreamFilter, IWorker {
public static TimeSpan parse_duration(string input) {
int days = 0, hours = 0, minutes = 0, seconds = 0;
@@ -42,28 +42,15 @@ namespace BCG {
: this(col_name, parse_duration(row_time_to_live)) {
}
public TimeBasedFilter(string rowTimeColumnName, TimeSpan rowTimeToLive)
: this(ViewStreamDatabase.instance, rowTimeColumnName, rowTimeToLive, new TimeSpan(0, 1, 0) /* 1 minute */) {
: this(rowTimeColumnName, rowTimeToLive, new TimeSpan(0, 1, 0) /* 1 minute */) {
}
public TimeBasedFilter(string rowTimeColumnName, TimeSpan rowTimeToLive, TimeSpan filterFrequency)
: this(ViewStreamDatabase.instance, rowTimeColumnName, rowTimeToLive, filterFrequency) {
}
public TimeBasedFilter(ViewStreamDatabase vsd, string rowTimeColumnName, TimeSpan rowTimeToLive)
: this(vsd, rowTimeColumnName, rowTimeToLive, new TimeSpan(0, 1, 0) /* 1 minute */) {
}
public TimeBasedFilter(ViewStreamDatabase vsd, string rowTimeColumnName, TimeSpan rowTimeToLive, TimeSpan filterFrequency) {
myVsd = vsd;
public TimeBasedFilter(string rowTimeColumnName, TimeSpan rowTimeToLive, TimeSpan filterFrequency) {
_time_col = rowTimeColumnName;
_ttl = rowTimeToLive;
myFilterFrequency = filterFrequency;
myNextFilter = DateTime.MinValue;
// add worker
myVsd.addWorker(this);
}
#endregion
@@ -97,29 +84,11 @@ namespace BCG {
}
#endregion
#region IDisposable
public void Dispose() {
// remove worker
myVsd.removeWorker(this);
}
#endregion
#region state
private ViewStreamDatabase myVsd;
private string _time_col;
private TimeSpan _ttl;
private TimeSpan myFilterFrequency;
private DateTime myNextFilter;
#endregion
}
[DataStreamUserRespresentation("timefilter")]
public class TimeBasedFilterStream : ViewStream {
public TimeBasedFilterStream(ViewStream base_stream, string time_col, string time_span)
:
this(ViewStreamDatabase.instance, base_stream, time_col, TimeBasedFilter.parse_duration(time_span)) { }
public TimeBasedFilterStream(ViewStreamDatabase vsd, ViewStream base_stream, string time_col, TimeSpan time_span) {
TimeBasedFilter filter = new TimeBasedFilter(vsd, time_col, time_span);
mySetRealSource(new FilterStream(base_stream, filter));
}
}
}
@@ -0,0 +1,14 @@
namespace BCG;
using System;
[DataStreamUserRespresentation("timefilter")]
public class TimeBasedFilterStream : ViewStream {
public TimeBasedFilterStream(ViewStream base_stream, string time_col, string time_span)
: this(base_stream, time_col, TimeBasedFilter.parse_duration(time_span)) { }
public TimeBasedFilterStream(ViewStream base_stream, string time_col, TimeSpan time_span) {
var filter = new TimeBasedFilter(time_col, time_span);
mySetRealSource(AttachStream<FilterStream>(new(base_stream, filter)));
}
}
@@ -5,8 +5,8 @@ namespace BCG {
public MaxStream(ViewStream stream, string outputCol, string inputCols) : this(stream, outputCol, inputCols.Split(' ')) {
}
public MaxStream(ViewStream stream, string outputCol, string[] inputCols) {
mySetRealSource(new GroupByStream(stream,
new MaxGroupByEvaluator(outputCol, inputCols)));
mySetRealSource(AttachStream(new GroupByStream(stream,
new MaxGroupByEvaluator(outputCol, inputCols))));
}
public MaxStream(ViewStream stream, string outputCol)
@@ -1,13 +1,15 @@
using System;
namespace BCG {
public class MinStream : ViewStream {
public MinStream(ViewStream stream, string outputCol, string[] inputCol) {
mySetRealSource(new GroupByStream(stream,
new MinGroupByEvaluator(outputCol, inputCol)));
}
public class MinStream : ViewStream {
public MinStream(ViewStream stream, string outputCol, string[] inputCol) {
mySetRealSource(AttachStream(new GroupByStream(stream,
new MinGroupByEvaluator(outputCol, inputCol))));
}
public MinStream(ViewStream stream, string outputCol)
: this(new AddColumnsStream(stream, "fakeCol=0"), outputCol, new string[] { "fakeCol" }) { }
}
public MinStream(ViewStream stream, string outputCol)
: this(new AddColumnsStream(stream, "fakeCol=0"), outputCol, new string[] { "fakeCol" }) {
ViewStream vs = AttachStream(new AddColumnsStream(stream, "fakeCol=0"));
vs = AttachStream(new GroupByStream(vs, new MinGroupByEvaluator(outputCol, new[] { "fakeCol" })));
mySetRealSource(vs);
}
}
}
@@ -26,22 +26,22 @@ namespace BCG {
public SumStream(ViewStream stream, string spaceSeparatedOutputCols, string spaceSeparatedGroupByCols) {
string[] outputCols = spaceSeparatedOutputCols.Split(' ');
string[] groupByCols = spaceSeparatedGroupByCols.Split(' ');
mySetRealSource(new GroupByStream(stream,
mySetRealSource(AttachStream(new GroupByStream(stream,
outputCols.Length == 1 ? new SumGroupByEvaluator(outputCols[0], groupByCols) as IGroupByEvaluator :
new MultiColumnSumGroupByEvaluator(spaceSeparatedOutputCols, spaceSeparatedGroupByCols)));
new MultiColumnSumGroupByEvaluator(spaceSeparatedOutputCols, spaceSeparatedGroupByCols))));
}
public SumStream(ViewStream stream, string[] output_columns) {
ViewStream rs = new AddColumnsStream(stream, "fake_col=0");
rs = new SumStream(rs, output_columns, new string[] { "fake_col" }, false);
rs = new ProjectStream(rs, "!fake_col");
ViewStream rs = AttachStream(new AddColumnsStream(stream, "fake_col=0"));
rs = AttachStream(new SumStream(rs, output_columns, new string[] { "fake_col" }, false));
rs = AttachStream(new ProjectStream(rs, "!fake_col"));
mySetRealSource(rs);
}
public SumStream(ViewStream stream, string space_seperated_output_columns, string space_seperated_groupby_columns, bool force_null_into_other_columns) {
string[] output_columns = space_seperated_output_columns.Split(' ');
string[] groupby_columns = String.IsNullOrEmpty(space_seperated_groupby_columns) ? new string[] { } : space_seperated_groupby_columns.Split(' ');
mySetRealSource(new SumStream(stream, output_columns, groupby_columns, force_null_into_other_columns));
mySetRealSource(AttachStream(new SumStream(stream, output_columns, groupby_columns, force_null_into_other_columns)));
}
public SumStream(ViewStream stream, string[] output_columns, string[] groupby_columns, bool force_null_into_other_columns) {
@@ -52,12 +52,13 @@ namespace BCG {
// rs = new GroupByStream(rs, new MultiColumnSumGroupByEvaluator(output_columns, new string[] { "fake_col" }));
// rs = new ProjectStream(rs, "!fake_col");
} else {
rs = new GroupByStream(stream, new MultiColumnSumGroupByEvaluator(output_columns, groupby_columns));
rs = AttachStream(new GroupByStream(stream, new MultiColumnSumGroupByEvaluator(output_columns, groupby_columns)));
}
if (force_null_into_other_columns) {
ArgumentNullException.ThrowIfNull(rs);
List<string> valid_cols = new List<string>(output_columns);
valid_cols.AddRange(groupby_columns);
rs = new MapStream(rs, new NullMapper(valid_cols.ToArray()));
rs = AttachStream(new MapStream(rs, new NullMapper(valid_cols.ToArray())));
}
mySetRealSource(rs);
}
@@ -71,9 +72,9 @@ namespace BCG {
public IntSumStream(ViewStream stream, string spaceSeparatedOutputCols, string spaceSeparatedGroupByCols) {
string[] outputCols = spaceSeparatedOutputCols.Split(' ');
string[] groupByCols = spaceSeparatedGroupByCols.Split(' ');
mySetRealSource(new GroupByStream(stream,
mySetRealSource(AttachStream(new GroupByStream(stream,
outputCols.Length == 1 ? new SumGroupByEvaluator<int>(outputCols[0], groupByCols) as IGroupByEvaluator :
new MultiColumnSumGroupByEvaluator<int>(spaceSeparatedOutputCols, spaceSeparatedGroupByCols)));
new MultiColumnSumGroupByEvaluator<int>(spaceSeparatedOutputCols, spaceSeparatedGroupByCols))));
}
}
@@ -82,8 +83,8 @@ namespace BCG {
public class SumStream<T> : ViewStream {
public SumStream(ViewStream stream, string outputCol, string spaceSeparatedInputCols) {
string[] inputCols = spaceSeparatedInputCols.Split(' ');
mySetRealSource(new GroupByStream(stream,
new SumGroupByEvaluator<T>(outputCol, inputCols)));
mySetRealSource(AttachStream(new GroupByStream(stream,
new SumGroupByEvaluator<T>(outputCol, inputCols))));
}
}
}
@@ -22,9 +22,9 @@ namespace BCG
{
if (myUseSuperJoin)
{
ViewStream rs = this is LeftJoinStream ?
new SuperLeftJoinStream(v1, joinCol1, cm1.columnExpression, v2, joinCol2, cm2.columnExpression) as ViewStream :
new SuperJoinStream(v1, joinCol1, cm1.columnExpression, v2, joinCol2, cm2.columnExpression) as ViewStream;
ViewStream rs = AttachStream(this is LeftJoinStream
? new SuperLeftJoinStream(v1, joinCol1, cm1.columnExpression, v2, joinCol2, cm2.columnExpression)
: new SuperJoinStream(v1, joinCol1, cm1.columnExpression, v2, joinCol2, cm2.columnExpression));
mySetRealSource(rs);
}
else
@@ -65,9 +65,9 @@ namespace BCG
{
if (myUseSuperJoin)
{
ViewStream rs = this is LeftJoinStream ?
new SuperLeftJoinStream(v1, joinCol1, v2, joinCol2) as ViewStream :
new SuperJoinStream(v1, joinCol1, v2, joinCol2) as ViewStream;
ViewStream rs = AttachStream(this is LeftJoinStream
? new SuperLeftJoinStream(v1, joinCol1, v2, joinCol2)
: new SuperJoinStream(v1, joinCol1, v2, joinCol2));
mySetRealSource(rs);
}
else
@@ -27,10 +27,10 @@ namespace BCG
rightJoinColArray = rightJoinColumns.Split(' ', ',', ';');
if (leftJoinColArray.Length == 1 && rightJoinColArray.Length == 1)
/// so that this handles regular joins too
mySetRealSource(leftJoin ? new SuperLeftJoinStream(leftStream, leftJoinColArray[0], leftProject,
mySetRealSource(AttachStream(leftJoin ? new SuperLeftJoinStream(leftStream, leftJoinColArray[0], leftProject,
rightStream, rightJoinColArray[0], rightProject) :
new SuperJoinStream(leftStream, leftJoinColArray[0], leftProject,
rightStream, rightJoinColArray[0], rightProject));
rightStream, rightJoinColArray[0], rightProject)));
else
{
string leftCols = "ljc:";
@@ -41,12 +41,12 @@ namespace BCG
foreach (string c in rightJoinColArray)
rightCols += c + " ";
rightCols = rightCols.Trim();
ViewStream res =
leftJoin ? new SuperLeftJoinStream(new ConcatenateColumnsStream(leftStream, leftCols), "ljc", leftProject,
new ConcatenateColumnsStream(rightStream, rightCols), "rjc", rightProject) :
new SuperJoinStream(new ConcatenateColumnsStream(leftStream, leftCols), "ljc", leftProject,
new ConcatenateColumnsStream(rightStream, rightCols), "rjc", rightProject);
res = new ProjectStream(res, "!ljc rjc");
leftStream = AttachStream(new ConcatenateColumnsStream(leftStream, leftCols));
rightStream = AttachStream(new ConcatenateColumnsStream(rightStream, rightCols));
ViewStream res = AttachStream(leftJoin
? new SuperLeftJoinStream(leftStream, "ljc", leftProject, rightStream, "rjc", rightProject)
: new SuperJoinStream(leftStream, "ljc", leftProject, rightStream, "rjc", rightProject));
res = AttachStream(new ProjectStream(res, "!ljc rjc"));
mySetRealSource(res);
}
}
@@ -15,7 +15,6 @@ namespace BCG
private readonly INonIncJoinInlineMapper _output_mapper;
private readonly BlueRowHasher _row_hasher;
private readonly TimeSpan _throttle_period;
private readonly ViewStreamDatabase _vsd;
private List<NonIncJoinBookkeeper> _book_keeping_data = new List<NonIncJoinBookkeeper>();
private IBlueTableSchema _output_schema = null;
@@ -35,22 +34,17 @@ namespace BCG
: this(NonIncJoinSpec.from_triplet(join_specs), null, 0)
{ }
public NonIncJoinStream(ViewStreamDatabase vsd, INonIncJoinInlineMapper output_mapper, int throttles_ms,
public NonIncJoinStream(INonIncJoinInlineMapper output_mapper, int throttles_ms,
params Triplet<ViewStream, string, string>[] join_specs)
: this(vsd, NonIncJoinSpec.from_triplet(join_specs), output_mapper, throttles_ms)
: this(NonIncJoinSpec.from_triplet(join_specs), output_mapper, throttles_ms)
{ }
public NonIncJoinStream(NonIncJoinSpec[] join_specs, INonIncJoinInlineMapper output_mapper, int throttles_ms = 0)
: this(ViewStreamDatabase.instance, join_specs, output_mapper, throttles_ms)
{ }
public NonIncJoinStream(ViewStreamDatabase vsd, NonIncJoinSpec[] join_specs, INonIncJoinInlineMapper output_mapper, int throttles_ms = 0)
{
_join_specs = join_specs;
_output_mapper = output_mapper;
_row_hasher = new BlueRowHasher(_join_specs[0].join_col_names.Length);
_throttle_period = throttles_ms == 0 ? TimeSpan.Zero : new TimeSpan(0, 0, 0, 0, throttles_ms);
_vsd = vsd;
}
protected override void myOnFirstSubscriber()
@@ -65,7 +59,7 @@ namespace BCG
}
if (_output_mapper != null)
_output_mapper.onMapperChange += _on_mapper_change;
_vsd.addWorker(this);
base.myOnFirstSubscriber();
}
protected override void myOnLastSubscriber()
@@ -74,7 +68,7 @@ namespace BCG
d.Dispose();
if (_output_mapper != null)
_output_mapper.onMapperChange -= _on_mapper_change;
_vsd.removeWorker(this);
base.myOnLastSubscriber();
}
private void _stream_callback()
@@ -52,7 +52,7 @@ namespace BCG {
throw (new Exception("Error occured at line: " + n.ToString() + " " + e.Message));
}
mySetRealSource(new BlueRowCollectionStaticStream(t));
mySetRealSource(AttachStream(new BlueRowCollectionStaticStream(t)));
}
public LiteralStream(string schema, string value) : this("literal", schema, value) { }
@@ -44,7 +44,7 @@ namespace BCG {
public MapStream(ViewStream vs, IStreamMapper mapper) : base(vs) {
if (myUseNewMap)
mySetRealSource(new SuperMapStream(vs, mapper));
mySetRealSource(AttachStream(new SuperMapStream(vs, mapper)));
else
myMapper = mapper;
}
@@ -57,7 +57,7 @@ namespace BCG {
public MapStream(ViewStream vs, IChangeableStreamMapper mapper) : this(vs, (IStreamMapper)mapper) {
if (myUseNewMap)
mySetRealSource(new SuperMapStream(vs, mapper));
mySetRealSource(AttachStream(new SuperMapStream(vs, mapper)));
else {
mapper.onMapperChange += new MapperChangeDelegate(myMapperChanged);
myChangeableMapper = mapper;
@@ -56,7 +56,6 @@ namespace bts {
}
[DataStreamUserRespresentation("dp_replay")]
public class DataPacketReplayStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
private readonly string _folder_path;
private readonly TimeSpan _fire_period;
IStreamMsgDeserilizer _deser = new StreamIODeserializationController();
@@ -66,24 +65,19 @@ namespace bts {
public DataPacketReplayStream(string folder_path, string fire_period)
:
this(ViewStreamDatabase.instance, folder_path, TimeBasedFilter.parse_duration(fire_period)) {
this(folder_path, TimeBasedFilter.parse_duration(fire_period)) {
}
public DataPacketReplayStream(ViewStreamDatabase vsd, string folder_path, TimeSpan fire_period) {
_vsd = vsd;
public DataPacketReplayStream(string folder_path, TimeSpan fire_period) {
_folder_path = folder_path;
_fire_period = fire_period;
}
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, _fire_period);
base.myOnFirstSubscriber();
_load_all_files();
}
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
public void doWork() {
if (_file_ptr >= _file_data.Count())
// we can set _file_ptr to 0 and keep going as well
@@ -139,4 +133,4 @@ namespace bts {
});
}
}
}
}
@@ -7,6 +7,7 @@ using System.Collections.Generic;
using System;
using System.IO;
using System.Linq;
using Microsoft.Extensions.DependencyInjection;
namespace bts {
public class PubSubTestingStream : ViewStream, IWorker {
@@ -61,20 +62,18 @@ namespace bts {
private readonly int _msg_size;
private readonly TimeSpan _publish_period;
private DateTime _last_publish_time = DateTime.MinValue;
private PublisherCore _pub_core = new PublisherCore();
private IPublisherCore? _pub_core_backing;
private BlueTable _table;
private int _n_pubs = 0;
private readonly int _max_pubs;
private int _n_updates = 0;
private bool _continue_sub_thread = true;
private ViewStreamDatabase _vsd;
private bool _fire_dirty = false;
private PayloadMessage _last_sent_msg, _last_recvd_msg;
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) {
_vsd = vsd;
public PubSubTestingStream(int publish_period_msecs, int max_pubs, int payload_size, string tag_name) {
_payload_sz = payload_size;
var schema = BlueDataUtility.form_schema("PubSubTest",
Tuple.Create("id", TypeCode.Int32),
@@ -93,18 +92,18 @@ namespace bts {
_tag_name = tag_name;
}
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) { }
IPublisherCore? _pub_core =>
_pub_core_backing ??= ActivatorUtilities.CreateInstance<IPublisherCore>(ServiceProvider);
protected override void myOnFirstSubscriber() {
_vsd.addWorker(this);
base.myOnFirstSubscriber();
System.Threading.Thread th = new System.Threading.Thread(_subscribe_thread_func);
th.IsBackground = true;
th.Start();
}
protected override void myOnLastSubscriber() {
_vsd.removeWorker(this);
base.myOnLastSubscriber();
_continue_sub_thread = false;
}
@@ -118,7 +117,7 @@ namespace bts {
private void _subscribe_thread_func() {
// subscription is done in another thread
SubscriberCore sub_core = new SubscriberCore();
var sub_core = ActivatorUtilities.CreateInstance<SubscriberCore>(ServiceProvider);
sub_core.subscribe(_tag_name);
while (_continue_sub_thread) {
// TODO recv_id
@@ -152,7 +151,7 @@ namespace bts {
}
var ser_bytes = _last_sent_msg.serialize();
_n_pubs++;
_pub_core.publish(String.Empty, _tag_name, ser_bytes);
Log.notnull(_pub_core).publish(string.Empty, _tag_name, ser_bytes);
_fire_dirty = true;
}
@@ -193,4 +192,4 @@ namespace bts {
}
}
}
}
@@ -1,202 +0,0 @@
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;
using Bts.PubSub.Routing;
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) {
_router = new PubSubRouter(new PubSubRouterConfig() {
PublisherRoutes = new() { new(server_url, ".*") },
SubscriberRoutes = new() { new(server_url, ".*") },
});
_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<IEnumerable<PubSubServerMessage>, 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();
}
}
}
@@ -25,7 +25,7 @@ namespace bts {
}
protected override void myOnFirstSubscriber() {
ViewStreamDatabase.instance.addWorker(this);
base.myOnFirstSubscriber();
_tcp_client.register_listener((msg) => { _on_msg(msg); });
_tcp_client.start();
}
@@ -55,4 +55,4 @@ namespace bts {
}
}
}
@@ -7,14 +7,14 @@ using System.Collections.Generic;
using MongoDB.Bson;
using BCG.Utilities.General;
using System.Linq;
using Bts.PubSub;
namespace bts {
// this class, a little specialized, takes the first snapshot thats fired
// from the base stream sorts them by a given column, and fires one row at a time
public class SnapshotReplayStream : ViewStream, IWorker {
public class SnapshotReplayStream : ViewStream, IParameterizedWorker {
private ViewStreamDatabase _vsd;
private ViewStream _snapshot_stream;
private readonly string _sort_col;
private readonly int _buffer_size;
@@ -25,22 +25,24 @@ namespace bts {
public SnapshotReplayStream(ViewStream base_str, string sort_col, int buffer_size, string fire_period)
: this(ViewStreamDatabase.instance, base_str, sort_col, buffer_size, TimeBasedFilter.parse_duration(fire_period)) { }
: this(base_str, sort_col, buffer_size, TimeBasedFilter.parse_duration(fire_period)) { }
public SnapshotReplayStream(ViewStreamDatabase vsd, ViewStream base_str, string sort_col, int buffer_size, TimeSpan fire_period) {
_vsd = vsd;
public SnapshotReplayStream(ViewStream base_str, string sort_col, int buffer_size, TimeSpan fire_period) {
_snapshot_stream = base_str;
_sort_col = sort_col;
_buffer_size = buffer_size;
_fire_period = fire_period;
}
TimeSpan IParameterizedWorker.Interval => _fire_period;
protected override void myOnFirstSubscriber() {
_snapshot_stream.onChange += _handle_stream_update;
MasterWorker.getInstance(false).addWorker(this, _fire_period);
base.myOnFirstSubscriber();
}
protected override void myOnLastSubscriber() {
_snapshot_stream.onChange -= _handle_stream_update;
base.myOnLastSubscriber();
}
@@ -74,4 +76,4 @@ namespace bts {
}
}
}
}
@@ -561,7 +561,7 @@ namespace BCG {
private static TestResult _run_pubsub_server_stream_tag_handler_test() {
// create message
var schema = create_schema_with_all_types();
var tag = "DS_" + schema.tableName;
var tag = DSTagManager.get_stream_tag(schema.tableName);
BlueTable cr_rows = new BlueTable(schema);
// 100 rows
int n_rows = 100;
@@ -570,7 +570,7 @@ namespace BCG {
cr_rows.addRow(nr);
}
PubsubServerSingleStreamTagHandler handler = new PubsubServerSingleStreamTagHandler();
PubsubServerSingleStreamTagHandler handler = new PubsubServerSingleStreamTagHandler(tag);
bts.StreamIOMessage.CreateMessage cr_msg = new bts.StreamIOMessage.CreateMessage(cr_rows);
bts.StreamIOMessage str_io_msg = new bts.StreamIOMessage(cr_msg);
IStreamIOMsgSerializer ser = new StreamIOSerializationController();
@@ -616,53 +616,52 @@ namespace BCG {
}
private static TestResult _run_single_tag_data_test_clean() {
// var tagHandlerFactory
// create message
var test_name = MethodBase.GetCurrentMethod()?.Name;
var schema = create_schema_with_all_types();
var tag = "DS_" + schema.tableName;
BlueTable table = new BlueTable(schema);
IStreamIOMsgSerializer serializer = new StreamIOSerializationController();
IStreamMsgDeserilizer deser = new StreamIODeserializationController();
bts.SingleTagData st_data = new bts.SingleTagData(tag, DSTagManager.DataStreamTagType, new PubsubServerStreamTagHandler());
// create with 100 rows
int n_rows = 100;
bool is_succ = true;
for (int i = 0; i < n_rows; i++) {
var nr = generate_random_row(schema, i);
// perform the same action SingletagData, and verify the states match
PublishMsg pub_msg = build_pub_msg(nr, null, table.Count == 0, ref serializer);
table.addRow(nr);
st_data.handle_publish(pub_msg);
var table_p = get_rows_from_single_tag_state(ref st_data, ref deser);
is_succ = is_succ && BlueDataUtility.compute_equality(table, table_p);
}
// change all rows
for (int i = 0; i < n_rows; i++) {
var nr = generate_random_row(schema, i);
var dr = table[i];
table.deleteRow(dr);
table.addRow(nr);
PublishMsg pub_msg = build_pub_msg(nr, dr, false, ref serializer);
st_data.handle_publish(pub_msg);
var table_p = get_rows_from_single_tag_state(ref st_data, ref deser);
is_succ = is_succ && BlueDataUtility.compute_equality(table, table_p);
}
// delete all rows
var aux_table = new BlueTable(table);
foreach (IBlueRow r in aux_table) {
table.deleteRow(r);
PublishMsg pub_msg = build_pub_msg(null, r, false, ref serializer);
st_data.handle_publish(pub_msg);
var table_p = get_rows_from_single_tag_state(ref st_data, ref deser);
is_succ = is_succ && BlueDataUtility.compute_equality(table, table_p);
}
return new TestResult() {
test_result = is_succ,
test_name = test_name
};
}
// private static TestResult _run_single_tag_data_test_clean() {
// // create message
// var test_name = MethodBase.GetCurrentMethod()?.Name;
// var schema = create_schema_with_all_types();
// var tag = "DS_" + schema.tableName;
// BlueTable table = new BlueTable(schema);
// IStreamIOMsgSerializer serializer = new StreamIOSerializationController();
// IStreamMsgDeserilizer deser = new StreamIODeserializationController();
// bts.SingleTagData st_data = new bts.SingleTagData(tag, DSTagManager.DataStreamTagType);
// // create with 100 rows
// int n_rows = 100;
// bool is_succ = true;
// for (int i = 0; i < n_rows; i++) {
// var nr = generate_random_row(schema, i);
// // perform the same action SingletagData, and verify the states match
// PublishMsg pub_msg = build_pub_msg(nr, null, table.Count == 0, ref serializer);
// table.addRow(nr);
// st_data.handle_publish(pub_msg);
// var table_p = get_rows_from_single_tag_state(ref st_data, ref deser);
// is_succ = is_succ && BlueDataUtility.compute_equality(table, table_p);
// }
// // change all rows
// for (int i = 0; i < n_rows; i++) {
// var nr = generate_random_row(schema, i);
// var dr = table[i];
// table.deleteRow(dr);
// table.addRow(nr);
// PublishMsg pub_msg = build_pub_msg(nr, dr, false, ref serializer);
// st_data.handle_publish(pub_msg);
// var table_p = get_rows_from_single_tag_state(ref st_data, ref deser);
// is_succ = is_succ && BlueDataUtility.compute_equality(table, table_p);
// }
// // delete all rows
// var aux_table = new BlueTable(table);
// foreach (IBlueRow r in aux_table) {
// table.deleteRow(r);
// PublishMsg pub_msg = build_pub_msg(null, r, false, ref serializer);
// st_data.handle_publish(pub_msg);
// var table_p = get_rows_from_single_tag_state(ref st_data, ref deser);
// is_succ = is_succ && BlueDataUtility.compute_equality(table, table_p);
// }
// return new TestResult() {
// test_result = is_succ,
// test_name = test_name
// };
// }
@@ -687,13 +686,12 @@ namespace BCG {
return rv;
}
public static IBlueRowCollection get_rows_from_single_tag_state(ref SingleTagData st_data, ref IStreamMsgDeserilizer deser) {
var content = st_data.get_current_content();
var msg_p = deser.deserialize(content);
Log.assert(msg_p.op == bts.StreamIOMessage.StreamOp.CREATE);
return msg_p.create_message.rows;
}
// public static IBlueRowCollection get_rows_from_single_tag_state(ref SingleTagData st_data, ref IStreamMsgDeserilizer deser) {
// var content = st_data.get_current_content();
// var msg_p = deser.deserialize(content);
// Log.assert(msg_p.op == bts.StreamIOMessage.StreamOp.CREATE);
// return msg_p.create_message.rows;
// }
public static PublishMsg build_pub_msg(IBlueRow added_row, IBlueRow removed_row, bool is_create,
ref IStreamIOMsgSerializer serializer) {
@@ -3,12 +3,11 @@ namespace BCG {
public class TickingTestStream : ViewStream {
public TickingTestStream(int ticking_period_msecs)
:
this(ViewStreamDatabase.instance, TimeSpan.FromSeconds(ticking_period_msecs)) { }
: this(TimeSpan.FromSeconds(ticking_period_msecs)) { }
public TickingTestStream(ViewStreamDatabase vsd, TimeSpan ticking_period_msecs) {
ViewStream vs = new DisplayableMillisecondStream(Convert.ToInt32(ticking_period_msecs.TotalMilliseconds), 10);
vs = new MathStream(vs, "x=id*123 y=x*x");
public TickingTestStream(TimeSpan ticking_period_msecs) {
ViewStream vs = AttachStream(new DisplayableMillisecondStream(Convert.ToInt32(ticking_period_msecs.TotalMilliseconds), 10));
vs = AttachStream(new MathStream(vs, "x=id*123 y=x*x"));
mySetRealSource(vs);
//vs.onChange += _on_str_update;
}
@@ -28,4 +27,4 @@ namespace BCG {
// myFireEvent(dp_out);
// }
}
}
}
@@ -6,6 +6,8 @@ using System.Security.Cryptography.X509Certificates;
using BCG;
using BCG.Utilities.General;
using Bts.PubSub;
using HtmlAgilityPack;
namespace BCG {
@@ -46,19 +48,17 @@ namespace BCG {
}
}
public class WebScapperStream : ViewStream, IWorker {
public class WebScapperStream : ViewStream, IParameterizedWorker {
private readonly string _url;
private readonly string _xpath;
private readonly ViewStreamDatabase _vsd;
private IDictionary<string, int> _line_id_map = new Dictionary<string, int>();
private readonly IBlueTableSchema _schema;
public WebScapperStream()
:
this("http://quotes.toscrape.com", "//span[@class='text']", ViewStreamDatabase.instance) { }
public WebScapperStream(string url, string xpath)
: this(url, xpath, ViewStreamDatabase.instance) { }
public WebScapperStream(string url, string xpath, ViewStreamDatabase vsd) {
this("http://quotes.toscrape.com", "//span[@class='text']") { }
public WebScapperStream(string url, string xpath) {
bts.Log.Gen.info($"WebScapperStream cons url: {url} xpath: {xpath}");
_schema = BlueDataUtility.form_schema($"{url}_{xpath}",
Tuple.Create("id", TypeCode.Int32),
@@ -66,16 +66,9 @@ namespace BCG {
Tuple.Create("line", TypeCode.String));
_url = url;
_xpath = xpath;
_vsd = vsd;
}
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(1));
}
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
TimeSpan IParameterizedWorker.Interval => TimeSpan.FromSeconds(1);
public void doWork() {
var lines = scrape_lines(_url, _xpath);
@@ -59,8 +59,8 @@ namespace bts {
// now we creaate a TimeFilter stream to keep the table as a circular buffer of the last N days
// passed A an argument in the constructor above
// time_filter is a string like 7d 1h 30 m 45s
output_stream = new TimeBasedFilterStream(output_stream, "created_at", ttl);
output_stream = AttachStream(new TimeBasedFilterStream(output_stream, "created_at", ttl));
mySetRealSource(output_stream);
}
}
}
}
@@ -58,7 +58,7 @@ public class MongoCollectionStream : ViewStream, IObservable<DataPacket> {
this.databaseName = databaseName;
this.url = url;
this.rowFactory = rowFactory;
mySetRealSource(new ObservableViewStream(this));
mySetRealSource(AttachStream(new ObservableViewStream(this)));
}
MongoClient client => clientBacking ??= new(url);
@@ -1,6 +1,8 @@
using BCG;
using BCG.Utilities.General;
using Bts.PubSub;
using MongoDB.Bson;
using MongoDB.Driver;
@@ -10,8 +12,7 @@ using System.Diagnostics;
namespace bts {
[DataStreamUserRespresentation("mongodb_sql")]
public class MongoDBSQLStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
public class MongoDBSQLStream : ViewStream, IParameterizedWorker {
private readonly string _connection_string, _db_name, _collection_name;
private MongoClient _client;
// make this a sorted table
@@ -23,28 +24,17 @@ namespace bts {
private IDictionary<string, int> _id_map = new Dictionary<string, int>();
// 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, ViewStreamDatabase vsd) {
public MongoDBSQLStream(string connection_string, string db_name, string collection_name) {
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 myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
TimeSpan IParameterizedWorker.Interval => TimeSpan.FromSeconds(10);
public void doWork() {
_read_mogodb();
@@ -107,8 +97,7 @@ namespace bts {
}
[DataStreamUserRespresentation("mongodb")]
public class MongoDBStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
public class MongoDBStream : ViewStream, IParameterizedWorker {
private readonly string _connection_string, _db_name, _collection_name;
private MongoClient _client;
private BlueTable _table;
@@ -118,28 +107,17 @@ namespace bts {
private IDictionary<string, int> _id_map = new Dictionary<string, int>();
// 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, ViewStreamDatabase vsd) {
public MongoDBStream(string connection_string, string db_name, string collection_name) {
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 myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
TimeSpan IParameterizedWorker.Interval => TimeSpan.FromSeconds(10);
public void doWork() {
_read_mogodb();
@@ -212,12 +190,12 @@ namespace bts {
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)) { }
this(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);
public MongoDBFlatStream(string connection_string, string db_name, string collection_name, string timestamp_col, TimeSpan ttl_duration) {
ViewStream rs = AttachStream(new MongoDBStream(connection_string, db_name, collection_name));
rs = AttachStream(new KeyValueToRelationalStream(rs, "doc_id", "key", "val"));
rs = AttachStream(new DateStandardizationStream(rs, timestamp_col, DateConversion.ISO8601));
//rs = new TimeBasedFilterStream(vsd, rs, timestamp_col, ttl_duration);
mySetRealSource(rs);
}
@@ -229,21 +207,21 @@ namespace bts {
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)) { }
this(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,
public NewsReplayStream(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,
r = AttachStream<MongoDBStream>(new(connection_string, db_name, collection_name));
r = AttachStream<KeyValueToRelationalStream>(new(r,
"doc_id",
"key",
"value");
r = new DateStandardizationStream(r,
"value"));
r = AttachStream<DateStandardizationStream>(new(r,
"created_at",
"ISO8601");
r = new SnapshotReplayStream(vsd, r, "created_at", buffer_sz, fire_period);
"ISO8601"));
r = AttachStream<SnapshotReplayStream>(new(r, "created_at", buffer_sz, fire_period));
mySetRealSource(r);
}
@@ -25,7 +25,7 @@ namespace BCG
myInColumn = inColumn;
myMuxedValues = (object[])muxedValues.Clone();
mySetRealSource(new MapStream(inStream, this));
mySetRealSource(AttachStream(new MapStream(inStream, this)));
}
#region IStreamMapper Members
@@ -8,14 +8,14 @@ namespace BCG
{
public NullToConstantStream(ViewStream v, string field, object constant)
{
if(string.IsNullOrEmpty(field))
mySetRealSource(new MapStream(v, new AllNullToConstantMapper(constant)));
else
mySetRealSource (new MapStream(v,new NullToConstantMapper(field, constant)));
var vs = AttachStream<MapStream>(string.IsNullOrEmpty(field)
? new(v, new AllNullToConstantMapper(constant))
: new(v,new NullToConstantMapper(field, constant)));
mySetRealSource(vs);
}
public NullToConstantStream(ViewStream v, object constant)
{
mySetRealSource(new MapStream(v, new AllNullToConstantMapper(constant)));
mySetRealSource(AttachStream(new MapStream(v, new AllNullToConstantMapper(constant))));
}
}
}
@@ -87,7 +87,7 @@ namespace BCG
public ChangeTableNameStream (ViewStream vs, string name)
{
myNewName = name;
mySetRealSource (new MapStream (vs, this));
mySetRealSource (AttachStream(new MapStream (vs, this)));
}
#region IStreamMapper Members
@@ -116,7 +116,7 @@ namespace bts {
public NewProjectStream(ViewStream vs, string project_expression) {
_stream = vs;
_project_expression = project_expression;
_stream = new MapStream(vs, this);
_stream = AttachStream(new MapStream(vs, this));
mySetRealSource(_stream);
}
@@ -30,7 +30,7 @@ namespace BCG {
public ProjectStream (ViewStream vs, string colExp, string tableName)
{
if (colExp == "!")
mySetRealSource (new ChangeTableNameStream (vs, tableName));
mySetRealSource (AttachStream(new ChangeTableNameStream (vs, tableName)));
else
{
myVs = vs;
@@ -134,7 +134,7 @@ namespace BCG
{
public CSVReaderStreamMapperStream(ViewStream stream, string output_schema)
{
mySetRealSource(new MapStream(stream, new CSVReaderStreamMapper(output_schema)));
mySetRealSource(AttachStream(new MapStream(stream, new CSVReaderStreamMapper(output_schema))));
}
}
@@ -24,6 +24,7 @@ namespace bts {
var dp_out = BlueDataUtility.build_dp_inc(_rows, myLastValue as DataPacket);
myFireEvent(dp_out);
}
@@ -11,13 +11,7 @@ namespace BCG
{
#region constructor
public TextWriterStream(ViewStream stream, string fileNamePrefix, int writeFreqSecs)
: this(ViewStreamDatabase.instance, stream, fileNamePrefix, writeFreqSecs)
{
}
public TextWriterStream(ViewStreamDatabase vsd, ViewStream stream, string fileNamePrefix, int writeFreqSecs)
{
myVsd = vsd;
myStream = stream;
myFileNamePrefix = fileNamePrefix;
myWriteFreqSecs = writeFreqSecs;
@@ -36,12 +30,12 @@ namespace BCG
protected override void myOnFirstSubscriber()
{
myStream.onChange +=new ChangeDelegate(myStreamChange);
myVsd.addWorker(this);
base.myOnFirstSubscriber();
}
protected override void myOnLastSubscriber()
{
myVsd.removeWorker(this);
base.myOnLastSubscriber();
myStream.onChange -= new ChangeDelegate(myStreamChange);
}
#endregion
@@ -16,20 +16,17 @@ namespace BCG
private ViewStream myParentStream;
private ThreadSafeQueue<DataPacket> myQueuedUpdates;
private ViewStreamDatabase myVSD;
public ThreadSafeStream(ViewStream parent) : this(ViewStreamDatabase.instance, parent) { }
public ThreadSafeStream(ViewStreamDatabase vsd, ViewStream parentStream)
public ThreadSafeStream(ViewStream parentStream)
{
myParentStream = parentStream;
myVSD = vsd;
myQueuedUpdates = new ThreadSafeQueue<DataPacket>();
}
protected override void myOnFirstSubscriber()
{
myParentStream.onChange += new ChangeDelegate(myParentStreamChangeHandler);
myVSD.addWorker(this);
base.myOnFirstSubscriber();
}
@@ -37,7 +34,7 @@ namespace BCG
protected override void myOnLastSubscriber()
{
myParentStream.onChange -= new ChangeDelegate(myParentStreamChangeHandler);
myVSD.removeWorker(this);
base.myOnLastSubscriber();
}
void myParentStreamChangeHandler(object newValue)
@@ -50,7 +50,7 @@ namespace BCG {
public DateStandardizationStream(ViewStream vs, string colName, DateConversion conversion) {
myConversionColName = colName;
myConversion = conversion;
mySetRealSource(new MapStream(vs, this));
mySetRealSource(AttachStream(new MapStream(vs, this)));
}
#region IStreamMapper Members
@@ -135,4 +135,4 @@ namespace BCG {
}
}
}
}
}
@@ -15,32 +15,23 @@ namespace BCG
this (offsetInDays,new TimeSpan(0))
{}
public DayStream (int offsetInDays, TimeSpan zeroH)
:
this (offsetInDays, zeroH, ViewStreamDatabase.instance)
{}
public DayStream (int offsetInDays, int h, int m, int s)
:
this (offsetInDays, new TimeSpan(0,h,m,s,0), ViewStreamDatabase.instance)
this (offsetInDays, new TimeSpan(0,h,m,s,0))
{}
public DayStream (int offsetInDays, TimeSpan zeroHour, ViewStreamDatabase db)
public DayStream (int offsetInDays, TimeSpan zeroHour)
{
myOffset = offsetInDays;
myZeroHour = zeroHour;
myDB = db;
}
protected override void myOnFirstSubscriber()
{
myDB.addWorker(this);
base.myOnFirstSubscriber();
myInitialize();
}
protected override void myOnLastSubscriber()
{myDB.removeWorker(this);}
public void doWork()
{myInitialize();}
@@ -19,7 +19,7 @@ namespace BCG
_src = input;
_time_stamp_field = time_stamp_field;
_latency_field = latency_field;
mySetRealSource(new MapStream(input, this));
mySetRealSource(AttachStream(new MapStream(input, this)));
}
#region IStreamMapper Members
@@ -1,57 +1,9 @@
using System;
using BCG.Utilities.General;
using bts;
namespace BCG {
[DataStreamUserRespresentation("milliseconds")]
public class MillisecondStream : ViewStream, IWorker {
private DateTime _last_fire_time = DateTime.MinValue;
private readonly TimeSpan _period;
private readonly int _size;
private readonly ViewStreamDatabase _vsd;
private IBlueTableSchema _schema;
private bts.CircularBufferTable _table;
public MillisecondStream(int interval_msecs, int size)
: this(ViewStreamDatabase.instance, interval_msecs, size) {
}
public MillisecondStream(ViewStreamDatabase vsd, int interval_msecs, int size) {
_vsd = vsd;
_period = TimeSpan.FromMilliseconds(interval_msecs);
_size = size;
_schema = BlueDataUtility.form_schema("Milliseconds",
Tuple.Create("id", TypeCode.Int32),
Tuple.Create("time", TypeCode.DateTime),
Tuple.Create("display_time", TypeCode.String));
_table = new CircularBufferTable(_schema, _size);
}
protected override void myOnFirstSubscriber() {
_vsd.addWorker(this);
}
protected override void myOnLastSubscriber() {
_vsd.removeWorker(this);
}
private int _id = 0;
public void doWork() {
if (DateTime.Now - _last_fire_time >= _period) {
_last_fire_time = DateTime.Now;
IBlueRow nr = new BlueRow(_schema, _id++, _last_fire_time,
_last_fire_time.ToString("yyyy-MM-dd HH:mm:ss.fff"));
_table.add_row(nr);
var dp = BlueDataUtility.build_dp_inc(_table, myLastValue as DataPacket);
myFireEvent(dp);
}
}
}
public class MSecondStream : TimeUnitStream {
public MSecondStream(int intervalInms, int size)
: base(DateTime.Now, new TimeSpan(0, 0, 0, 0, intervalInms), size, ViewStreamDatabase.instance) { }
: base(DateTime.Now, new TimeSpan(0, 0, 0, 0, intervalInms), size) { }
protected override DateTime myRoundTime(DateTime dt) { return dt; }
}
@@ -61,7 +13,9 @@ namespace BCG {
new ColumnInfo("id", 0, typeof(int)),
new ColumnInfo("time", 1, typeof(string)));
public DisplayableMillisecondStream(int intervalInMs, int sz) {
mySetRealSource(new MapStream(new MSecondStream(intervalInMs, sz), this));
ViewStream vs = AttachStream(new MSecondStream(intervalInMs, sz));
vs = AttachStream(new MapStream(vs, this));
mySetRealSource(vs);
}
#region IStreamMapper Members
@@ -4,22 +4,14 @@ namespace BCG
{
public class SecondStream : TimeUnitStream
{
public SecondStream (DateTime zeroTime, int intervalInSeconds, int sizeOfStream, ViewStreamDatabase db) :
base (zeroTime, new TimeSpan (0,0,0,intervalInSeconds,0), sizeOfStream, db)
{}
public SecondStream (DateTime zeroTime, int intervalInSeconds, int sizeOfStream) :
this (zeroTime, intervalInSeconds, sizeOfStream, ViewStreamDatabase.instance)
base (zeroTime, new TimeSpan (0,0,0,intervalInSeconds,0), sizeOfStream)
{}
public SecondStream (int intervalInSeconds, int size)
: this (DateTime.Now, intervalInSeconds, size)
{}
public SecondStream (DateTime zeroTime, int sizeOfStream, ViewStreamDatabase db) :
this (zeroTime, 1, sizeOfStream, db)
{}
public SecondStream (DateTime zeroTime, int sizeOfStream) :
this (zeroTime, 1, sizeOfStream)
{}
@@ -10,36 +10,21 @@ namespace BCG
public class TickStream : ViewStream, IWorker
{
private DateTime myLastFireTime = DateTime.Now;
private ViewStreamDatabase myVsd;
private TimeSpan myPeriod;
private BlueTable myTable = new BlueTable(new BlueTableSchema("Ticks",
new ColumnInfo("id", 0, typeof(int)),
new ColumnInfo("ticks", 1, typeof(int))));
public TickStream()
: this(ViewStreamDatabase.instance, 0)
{ }
public TickStream(int periodMs)
: this(ViewStreamDatabase.instance, periodMs)
: this(0)
{ }
public TickStream(ViewStreamDatabase vsd, int periodMs)
public TickStream(int periodMs)
{
myVsd = vsd;
myPeriod = new TimeSpan(0, 0, 0, 0, periodMs);
}
protected override void myOnFirstSubscriber()
{
myVsd.addWorker(this);
}
protected override void myOnLastSubscriber()
{
myVsd.removeWorker(this);
}
#region IWorker Members
public void doWork()
@@ -2,6 +2,8 @@ using System;
using BCG.Utilities.General;
using Bts.DataStreamCore;
namespace BCG {
public abstract class TimeUnitStream : ViewStream, IWorker {
protected TimeSpan myOffset;
@@ -9,15 +11,13 @@ namespace BCG {
protected int mySizeOfStream;
protected DateTime myZeroTime;
protected CircularRowCollection myCollection;
protected ViewStreamDatabase myDb;
protected int myId = 0;
public TimeUnitStream(DateTime zeroTime, TimeSpan offSet, int sizeOfStream, ViewStreamDatabase db) {
public TimeUnitStream(DateTime zeroTime, TimeSpan offSet, int sizeOfStream) {
myConstructionTime = DateTime.Now;
myZeroTime = zeroTime;
myOffset = offSet;
mySizeOfStream = sizeOfStream;
myDb = db;
myCollection = new CircularRowCollection(
new BlueTableSchema("Time",
new ColumnInfo("id", 0, typeof(Int32)), new ColumnInfo("time", 1, typeof(DateTime))),
@@ -27,17 +27,15 @@ namespace BCG {
#region ViewStream overriden methods
protected override void myOnFirstSubscriber() {
myDb.addWorker(this);
myInit();
}
protected override void myOnLastSubscriber() {
myDb.removeWorker(this);
}
#endregion
protected virtual void myInit() {
if (!this.HasSubscribers()) {
return;
}
int periods = (int)
(DateTime.Now - myConstructionTime).TotalMilliseconds / (int)myOffset.TotalMilliseconds;
@@ -15,7 +15,7 @@ namespace BCG
public AppendStream(ViewStream v1, ViewStream v2)
{
if (myUseNewAppend)
mySetRealSource(new MutableAppendStream(v1, v2));
mySetRealSource(AttachStream(new MutableAppendStream(v1, v2)));
else
{
myLeftVs = v1;
@@ -27,7 +27,7 @@ namespace BCG
{
if (myUseNewAppend)
mySetRealSource(new MutableAppendStream(vss));
mySetRealSource(AttachStream(new MutableAppendStream(vss)));
else
{
if (vss.Length == 1)
@@ -36,7 +36,7 @@ namespace BCG
{
AppendStream rv = null;
foreach (ViewStream s in vss)
rv = rv == null ? new AppendStream(new ViewStream[] { s }) : new AppendStream(rv, s);
rv = AttachStream<AppendStream>(rv is null ? new(s) : new(rv, s));
mySetRealSource(rv);
}
else
@@ -112,7 +112,9 @@ namespace BCG
{
public MergeStream (ViewStream v1, ViewStream v2, string[] mergeCols)
{
mySetRealSource (new DistinctRowStream (new AppendStream (v1, v2), mergeCols));
ViewStream vs = AttachStream(new AppendStream (v1, v2));
vs = AttachStream(new DistinctRowStream (vs, mergeCols));
mySetRealSource(vs);
}
}
}
@@ -21,7 +21,7 @@ namespace BCG
{
AppendStreamModified rv = null;
foreach (ViewStream s in vss)
rv = rv == null ? new AppendStreamModified (new ViewStream[] {s}) : new AppendStreamModified (rv, s);
rv = AttachStream<AppendStreamModified>(rv is null ? new (new[] {s}) : new (rv, s));
mySetRealSource (rv);
}
}
@@ -1,5 +1,3 @@
using Bts.DataStreamCore;
using System;
namespace BCG {
@@ -9,15 +7,17 @@ namespace BCG {
/// </summary>
public class CircularPublisher {
private CircularRowCollection myBuffer;
private readonly IPubTableBroker myBroker;
private PubTableBroker myBroker;
private int myBufferSize;
public CircularPublisher(int bufferSize, IPubTableBroker broker) {
readonly IPublisherCore publisherCore;
public CircularPublisher(int bufferSize, IPublisherCore publisherCore) {
myBufferSize = bufferSize;
myBroker = broker;
this.publisherCore = publisherCore;
}
public CircularPublisher(int bufferSize, IPubTableBroker broker, IBlueRowCollection rows)
: this(bufferSize, broker) {
public CircularPublisher(int bufferSize, IPublisherCore publisherCore, IBlueRowCollection rows)
: this(bufferSize, publisherCore) {
publishRows(rows);
}
@@ -26,6 +26,7 @@ namespace BCG {
myBuffer = new CircularRowCollection(
new BlueTableSchema(nr.schema.tableName, nr.schema.columnInfo),
myBufferSize);
myBroker = new PubTableBroker(publisherCore);
myBroker.create_table(myBuffer);
}
IBlueRow br = myBuffer.addRowWithReturn(nr);
@@ -40,6 +41,7 @@ namespace BCG {
new BlueTableSchema(BlueDataUtility.extractRow(nrs, 0).schema.tableName,
BlueDataUtility.extractRow(nrs, 0).schema.columnInfo),
myBufferSize);
myBroker = new PubTableBroker(publisherCore);
myBroker.create_table(myBuffer);
}
BlueTable deletes = new BlueTable(myBuffer.schema);
@@ -1,18 +1,31 @@
namespace BCG {
using Bts.DataStreamCore;
using Microsoft.Extensions.DependencyInjection;
using System;
// this for testing purposes mainly, its wraps a ViewPublisher around a stream,
// and hence epublishes it out to the PS
public class PublisherStream : ViewStream {
private readonly ViewStream _main_stream;
private readonly string _name;
private ViewPublisher _vs_publisher;
private readonly PubTableBroker _broker = new PubTableBroker();
IPubTableBroker? _broker_backing;
public PublisherStream(ViewStream main_stream, string name) {
_main_stream = main_stream;
_name = name;
}
IPubTableBroker? _broker =>
_broker_backing ??= ServiceProvider?.GetService<IPubTableBroker>();
protected override void myOnFirstSubscriber() {
if (_broker is null) {
throw new InvalidOperationException(
"Unable to initialize PublisherStream because IPubTableBroker is null");
}
_vs_publisher = new ViewPublisher(_name, _broker, _main_stream);
_main_stream.onChange += new DataStream.ChangeDelegate(_handle_update);
}
@@ -28,4 +41,4 @@ namespace BCG {
}
}
}
@@ -8,7 +8,6 @@ using Bts.DataStreamCore;
namespace BCG {
public class ViewPublisher : IDisposable {
protected ViewPublisher() { }
private readonly IPubTableBroker _broker;
protected readonly ViewStream _stream;
private int _n_updates = 0;
@@ -1,11 +1,94 @@
namespace Bts.DataStream.Services;
using BCG;
using BCG.Utilities.General;
using bts.api.core;
using Bts.PubSub;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
public static class DataStreamServices {
public static IServiceCollection AddRemoteQueryManager(this IServiceCollection services) =>
services.AddSingleton<IWorker, RemoteQueryManager>();
}
using System;
using System.Linq;
// public static class DataStreamPublisherServices {
// /// <summary>
// /// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
// /// already.
// /// </summary>
// /// <typeparam name="T"></typeparam>
// /// <param name="services"></param>
// /// <returns></returns>
// public static IServiceCollection EnsureSingleton<T>(this IServiceCollection services) where T : class =>
// services.HasService<T>()
// ? services
// : services.AddSingleton<T>();
// /// <summary>
// /// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
// /// already.
// /// </summary>
// /// <typeparam name="T"></typeparam>
// /// <param name="services"></param>
// /// <returns></returns>
// public static IServiceCollection EnsureSingleton<T>(this IServiceCollection services, Func<IServiceProvider, T> implementationFactory) where T : class =>
// services.HasService<T>()
// ? services
// : services.AddSingleton(implementationFactory);
// /// <summary>
// /// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
// /// already.
// /// </summary>
// /// <typeparam name="T">The type of the service to add.</typeparam>
// /// <typeparam name="TImplementation">The type of the implementation to use.</typeparam>
// /// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
// /// <returns>A reference to this instance after the operation has completed.</returns>
// public static IServiceCollection EnsureSingleton<T, TImplementation>(this IServiceCollection services) where T : class where TImplementation : class, T =>
// services.HasService<T>()
// ? services
// : services.AddSingleton<T, TImplementation>();
// /// <summary>
// /// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
// /// already.
// /// </summary>
// /// <typeparam name="T">The type of the service to add.</typeparam>
// /// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
// /// <returns>A reference to this instance after the operation has completed.</returns>
// public static IServiceCollection EnsureKeyedSingleton<T>(this IServiceCollection services, object? key) where T : class =>
// services.HasKeyedService<T>(key)
// ? services
// : services.AddKeyedSingleton<T>(key);
// /// <summary>
// /// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
// /// already.
// /// </summary>
// /// <typeparam name="T">The type of the service to add.</typeparam>
// /// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
// /// <returns>A reference to this instance after the operation has completed.</returns>
// public static IServiceCollection EnsureHostedService<T>(this IServiceCollection services) where T : class, IHostedService =>
// services.HasService<T>()
// ? services
// : services.AddHostedService<T>();
// /// <summary>
// /// Adds a singleton service only if a service of type <typeparamref name="T"/> hasn't been added
// /// already.
// /// </summary>
// /// <typeparam name="T">The type of the service to add.</typeparam>
// /// <param name="services">The <see cref="IServiceCollection"/> to add the service to.</param>
// /// <returns>A reference to this instance after the operation has completed.</returns>
// public static IServiceCollection EnsureHostedService<T>(this IServiceCollection services, Func<IServiceProvider, T> implementationFactory) where T : class, IHostedService =>
// services.HasService<T>()
// ? services
// : services.AddHostedService(implementationFactory);
// public static bool HasService<T>(this IServiceCollection services) =>
// services.Any(s => s.ServiceType.IsAssignableTo(typeof(T)));
// public static bool HasKeyedService<T>(this IServiceCollection services, object? key) =>
// services.Any(s => Equals(s.ServiceKey, key) && s.ServiceType.IsAssignableTo(typeof(T)));
// }
@@ -16,6 +16,12 @@ namespace bts {
_buffer_sz = buffer_sz;
}
// public OrderedBlueTable(List<IBlueRow> rows, int buffer_sz) {
// _buffer_sz = buffer_sz;
// foreach (IBlueRow r in rows)
// add_row(r);
// }
// returns the row that might need to be deleted to keep
// the circular buffer intact
public IBlueRow add_row(IBlueRow row) {
@@ -0,0 +1,7 @@
namespace Bts.DataStreamCore;
using BCG;
public interface IConfiguredViewStream {
// ViewStream
}
@@ -17,8 +17,6 @@ namespace BCG
{
internal static event EventHandler onFireStart;
protected bool myTestMode = Configuration.get<DataStreamConfig>().IsTestMode;
protected bool myFirstSubImplemented = true,
myLastSubImplemented = true;
protected IDictionary myDelMap = new Hashtable ();
private Guid myID = Guid.Empty;
public DateTime last_fired_time = DateTime.MinValue;
@@ -123,8 +121,8 @@ namespace BCG
protected void mySetRealSource (DataStream src)
{myRealSource = src;}
protected virtual void myOnFirstSubscriber() {myFirstSubImplemented = false;}
protected virtual void myOnLastSubscriber () {myLastSubImplemented = false;}
protected virtual void myOnFirstSubscriber() { }
protected virtual void myOnLastSubscriber () { }
protected virtual void myFireEvent(object o)
{
last_fired_time = DateTime.Now;
@@ -14,6 +14,7 @@
<ItemGroup>
<ProjectReference Include="..\PubSub\PubSub.csproj" />
<ProjectReference Include="..\..\client\spark_core\spark_core.csproj" />
<ProjectReference Include="..\Bts.Extensions.DependencyInjection\Bts.Extensions.DependencyInjection.csproj" />
</ItemGroup>
<ItemGroup>
@@ -0,0 +1,8 @@
namespace Bts.DataStreamCore;
using BCG;
public static class DataStreamCoreExtensions {
public static bool HasSubscribers(this DataStream dataStream) =>
dataStream.numberOfSubscribers > 0;
}
@@ -1,27 +1,52 @@
#nullable enable
namespace BCG {
using Bts.DataStream;
using bts;
using Bts.DataStream;
using Bts.Extensions.DependencyInjection;
using mbts.config;
using System;
using System.Collections;
using System.Reflection;
using System.IO;
using System.Collections.Generic;
using Bts.DataStreamCore;
using Bts.PubSub;
public class PrimitiveManager {
private IDictionary myTypeLookupTable = new Hashtable();
private static PrimitiveManager mySelf;
Dictionary<string, Type> myTypeLookupTable = new();
private ArrayList myAssemblyNames = new ArrayList();
readonly IPubTableBroker? broker;
readonly IPubSubRouter? pubSubRouter;
readonly IPublisherCore? publisherCore;
readonly ISubscriberCore? subscriberCore;
readonly ViewStreamDatabase? viewStreamDatabase;
readonly IWorkScheduler? defaultWorkScheduler;
readonly IServiceProvider serviceProvider;
private PrimitiveManager() {
public PrimitiveManager(
IPubTableBroker? broker = null,
IPubSubRouter? pubSubRouter = null,
IPublisherCore? publisherCore = null,
ISubscriberCore? subscriberCore = null,
ViewStreamDatabase? viewStreamDatabase = null,
IWorkScheduler? defaultWorkScheduler = null
) {
this.broker = broker;
this.pubSubRouter = pubSubRouter;
this.publisherCore = publisherCore;
this.subscriberCore = subscriberCore;
this.viewStreamDatabase = viewStreamDatabase;
this.defaultWorkScheduler = defaultWorkScheduler;
myLoadAllTypes();
}
private void myLoadAllTypes() {
AppDomain currentDomain = AppDomain.CurrentDomain;
currentDomain.AssemblyResolve += new ResolveEventHandler(myAssemblyResolver);
loadAssembly(this.GetType().Assembly.FullName);
currentDomain.AssemblyResolve += myAssemblyResolver;
loadAssembly(GetType()!.Assembly!.FullName!);
var assemblies = new string[] { "DataStream.dll", "SharedStreams.dll" };
var config_assemblies = Configuration.get<DataStreamConfig>().Assemblies;
bts.Log.Gen.info($"PrimitiveManager.myLoadAllTypes config_assemblies: {config_assemblies}");
@@ -35,10 +60,12 @@ namespace BCG {
}
}
private static Assembly myAssemblyResolver(object sender, ResolveEventArgs args) {
foreach (Assembly a in AppDomain.CurrentDomain.GetAssemblies())
if (args.Name.Equals(a.FullName))
static Assembly? myAssemblyResolver(object? sender, ResolveEventArgs args) {
foreach (var a in AppDomain.CurrentDomain.GetAssemblies()) {
if (args.Name.Equals(a.FullName)) {
return a;
}
}
return null;
}
@@ -51,34 +78,34 @@ namespace BCG {
if (myAssemblyNames.Contains(assemblyName))
return;
myAssemblyNames.Add(assemblyName);
Assembly loadedAssembly = GetType().Assembly.FullName.Equals(assemblyName) ?
Assembly loadedAssembly = GetType()!.Assembly!.FullName!.Equals(assemblyName) ?
GetType().Assembly :
Assembly.LoadFrom(assemblyName);
foreach (Type type in loadedAssembly.GetTypes())
foreach (Type type in loadedAssembly.GetTypes()) {
if (!type.IsInterface
&& !type.IsAbstract
&& type.IsSubclassOf(typeof(ViewStream))
&& type.IsPublic) {
bool isVisible = true;
string userRep = createDefaultName(type.Name);
foreach (Attribute att in type.GetCustomAttributes(false))
if (att is DataStreamUserRespresentation) {
DataStreamUserRespresentation dsatt = att as DataStreamUserRespresentation;
isVisible = dsatt.isVisible;
userRep = isVisible ?
dsatt.friendlyName == null ? userRep : dsatt.friendlyName
: null;
break;
}
string? userRep = createDefaultName(type.Name);
var dsatt = type.GetCustomAttribute<DataStreamUserRespresentation>();
if (dsatt is not null) {
isVisible = dsatt.isVisible;
userRep = isVisible ?
dsatt.friendlyName is null ? userRep : dsatt.friendlyName
: null;
}
if (isVisible) {
Log.assert_d(myTypeLookupTable[userRep] == null, type.FullName);
Log.notnull(userRep);
Log.assert_d(!myTypeLookupTable.ContainsKey(userRep), type.FullName!);
myTypeLookupTable[userRep] = type;
}
}
}
}
public static string createDefaultName(string typeName) {
static string createDefaultName(string typeName) {
if (!typeName.EndsWith("Stream"))
return typeName;
string retVal = typeName.Remove(typeName.Length - "Stream".Length, "Stream".Length);
@@ -88,22 +115,14 @@ namespace BCG {
return retVal;
}
public static PrimitiveManager instance {
get { return mySelf = mySelf == null ? new PrimitiveManager() : mySelf; }
}
/// <summary>
/// Returns the Type name corresponding to the primitive name
/// </summary>
/// <param name="friendlyName"></param>
/// <returns></returns>
public string find(string friendlyName) {
if (myTypeLookupTable[friendlyName] == null)
throw new ApplicationException(friendlyName + " is not the friendly name of any of the loaded types");
return (myTypeLookupTable[friendlyName] as Type).Name;
}
public Type find(string friendlyName) =>
myTypeLookupTable.TryGetValue(friendlyName, out var type) ? type
: throw new ApplicationException($"'{friendlyName}' is not the friendly name of any of the loaded types");
/// <summary>
/// Constructs and returns a ViewStream of type corresponding to the friendly name
@@ -112,18 +131,31 @@ namespace BCG {
/// <param name="args">args can be null or an empty ICollection if the paramterless constructor
/// is the desired one</param>
/// <returns></returns>
public ViewStream build(string friendlyName, ICollection args) {
object[] constructorArgs = null;
public ViewStream? build(string friendlyName, ICollection args) {
object[]? constructorArgs = null;
if (args != null && args.Count > 0) {
constructorArgs = new object[args.Count];
args.CopyTo(constructorArgs, 0);
}
if (myTypeLookupTable[friendlyName] == null)
throw new ApplicationException(friendlyName + " is not the friendly name of any of the loaded types");
return
constructorArgs == null ?
Activator.CreateInstance(myTypeLookupTable[friendlyName] as Type) as ViewStream :
Activator.CreateInstance(myTypeLookupTable[friendlyName] as Type, constructorArgs) as ViewStream;
var type = find(friendlyName);
var vs = constructorArgs == null ?
Activator.CreateInstance(type) as ViewStream :
Activator.CreateInstance(type, constructorArgs) as ViewStream;
if (vs is not null) {
InitializeViewStream(vs);
}
return vs;
}
public void InitializeViewStream(ViewStream viewStream) {
viewStream.Reset(serviceProvider);
// (viewStream as IDependOn<IPublisherCore>)?.Reset(publisherCore);
// (viewStream as IDependOn<ISubscriberCore>)?.Reset(subscriberCore);
// (viewStream as IDependOn<IPubTableBroker>)?.Reset(broker);
// (viewStream as IDependOn<IPubSubRouter>)?.Reset(pubSubRouter);
// (viewStream as IDependOn<ViewStreamDatabase>)?.Reset(viewStreamDatabase);
// (viewStream as IDependOn<IViewStreamDatabase>)?.Reset(viewStreamDatabase);
// (viewStream as IDependOn<IWorkScheduler>)?.Reset(defaultWorkScheduler);
}
/// <summary>
@@ -0,0 +1,7 @@
namespace Bts.DataStreamCore;
using BCG.Utilities.General;
using Bts.PubSub;
public interface IViewStreamDatabase : IWorker, IWorkScheduler { }
@@ -0,0 +1,13 @@
#nullable enable
namespace Bts.DataStreamCore;
using BCG.Utilities.General;
using System;
/// <summary>
/// Implementations of <see cref="BCG.ViewStream"/> can implement this interface to
/// perform
/// </summary>
public interface IViewStreamWorker : IWorker {
}
@@ -21,9 +21,6 @@ namespace BCG {
protected bool myWaitingForFirstOp = false;
public PubTableBroker()
: this(new PublisherCore()) { }
public PubTableBroker(IPublisherCore publisher) {
_publisher_core = publisher;
Log.Gen.info("PubTableBroker constructed");
@@ -53,13 +50,13 @@ namespace BCG {
public virtual void make_batch_changes(IBlueRowCollection new_rows,
IBlueRowCollection del_rows,
IBlueRowCollection added_rows) {
// string log_msg = $"PubTableBroker.make_batch_changes table: {new_rows.schema.tableName} del: {del_rows.Count} added: {added_rows.Count}";
// foreach (IBlueRow r in del_rows)
// log_msg += $"id: {r["id"]} ";
// log_msg += "addedRows: ";
// foreach (IBlueRow r in added_rows)
// log_msg += $"id: {r["id"]}";
// bts.Log.Gen.info(log_msg);
string log_msg = $"PubTableBroker.make_batch_changes table: {new_rows.schema.tableName} del: {del_rows.Count} added: {added_rows.Count}";
foreach (IBlueRow r in del_rows)
log_msg += $"id: {r["id"]} ";
log_msg += "addedRows: ";
foreach (IBlueRow r in added_rows)
log_msg += $"id: {r["id"]}";
bts.Log.Gen.info(log_msg);
SinglePubTableBroker table_publisher = null;
var name_key = new_rows.schema.tableName.ToLower();
if (!_publisher_map.TryGetValue(name_key, out table_publisher))
@@ -5,21 +5,18 @@ using System.Linq;
using bts;
namespace BCG
{
namespace BCG {
// This class is responsible for mantaining the state fo a table. It gets the
// bytes from the pubsubserver via the handle_updates call (called by VSD), deserializes
// updates the state of the table, and fires the TableUpdateDelegate.
public class TableAccumulator2 : ITableAccumulator
{
public class TableAccumulator2 : ITableAccumulator {
private readonly string _tag;
private readonly ISubscriberCore _subscriber_core;
IStreamMsgDeserilizer _deser = new StreamIODeserializationController();
private BlueTable _table;
private TableUpdateDelegate _on_update_delegate;
public TableAccumulator2(string tag, ISubscriberCore sub_core)
{
public TableAccumulator2(string tag, ISubscriberCore sub_core) {
_tag = tag;
_subscriber_core = sub_core;
}
@@ -28,13 +25,10 @@ namespace BCG
//public bool is_active() { return _on_update_delegate != null; }
public event TableUpdateDelegate on_update
{
add
{
public event TableUpdateDelegate on_update {
add {
Log.Gen.info("TableAccumulator2.on_update.add {_tag} subscribed");
if (_on_update_delegate == null)
{
if (_on_update_delegate == null) {
//ie first subscription
Log.Gen.info("TableAccumulator2.on_update {_tag} subscribed for the first time");
_subscriber_core.subscribe(_tag);
@@ -44,17 +38,14 @@ namespace BCG
if (_table != null)
value(_table, _table, new BlueTable(_table.schema));
}
remove
{
remove {
Log.Gen.info("TableAccumulator2.on_update.remove {_tag} unsubscribed");
if (_on_update_delegate != null)
{
if (_on_update_delegate != null) {
_on_update_delegate =
Delegate.Remove(_on_update_delegate, value) as TableUpdateDelegate;
int n_subs = _on_update_delegate == null ? 0 : _on_update_delegate.GetInvocationList().Length;
Log.Gen.info("TableAccumulator2.on_update.remove {_tag} unsubscribed n_subs: {n_subs}");
if (_on_update_delegate == null)
{
if (_on_update_delegate == null) {
_subscriber_core.unsubscribe(_tag);
// TODO
//_table = null;
@@ -64,19 +55,16 @@ namespace BCG
}
}
private int _update_no = 0;
public void handle_updates(IEnumerable<byte[]> msgs)
{
public void handle_updates(IEnumerable<byte[]> msgs) {
if (msgs == null || msgs.Count() == 0)
return;
Log.Gen.info($"TableAccumulator2.handle_updates _update_no: {_update_no++} msgs: {msgs.Count()} msg_0: {msgs.ElementAt(0).Length}");
IBlueRowCollection table_0 = _table == null ? null : new BlueTable(_table);
foreach (var msg in msgs)
{
foreach (var msg in msgs) {
bts.StreamIOMessage str_msg = _deser.deserialize(msg);
var state_txt = _table == null ? "NULL" : _table.Count.ToString();// BlueDataUtility.id_string(_table);
var state_txt = _table == null ? "NULL" : BlueDataUtility.id_string(_table);
Log.Gen.info($"TableAccumulator2.handle_update tag: {_tag} sz: {msg.Length} str_msg: {str_msg} _state: {state_txt}");
switch (str_msg.op)
{
switch (str_msg.op) {
case bts.StreamIOMessage.StreamOp.CREATE:
_table = new BlueTable(str_msg.create_message.rows);
break;
@@ -89,13 +77,11 @@ namespace BCG
// this could be made faster if we just do this on the CREATE message, and pass-through
// what we get on the MULTIPLE_REMOVE_ADD
IBlueRowCollection d_rows, a_rows;
//BlueDataUtility.compute_differences_id_based(_table, table_0, out d_rows, out a_rows);
BlueDataUtility.compute_differences_fast(_table, table_0, out d_rows, out a_rows);
BlueDataUtility.compute_differences_id_based(_table, table_0, out d_rows, out a_rows);
_on_update_delegate(_table, a_rows, d_rows);
}
public void update(Stream[] streams)
{
public void update(Stream[] streams) {
}
#endregion
@@ -95,8 +95,8 @@ namespace bts {
}
public static IBlueTableSchema deserialize_schema(ref BinaryReader br) {
// var hex_txt = to_hex(br.BaseStream);
// bts.Log.Gen.info($"StreamSerializationUtils.deserialize_schema bytes: {hex_txt}");
var hex_txt = to_hex(br.BaseStream);
bts.Log.Gen.info($"StreamSerializationUtils.deserialize_schema bytes: {hex_txt}");
string table_name = br.read_string_cpp();
int n_cols = br.ReadInt32(); // doing a br.Read was only advancing the data 1 byte!!
List<ColumnInfo> cis = new List<ColumnInfo>();
@@ -137,8 +137,8 @@ namespace bts {
public static ColumnInfo deserialize_column_info(ref BinaryReader br) {
//col_no, name, type>
// var hex_txt = to_hex(br.BaseStream);
// bts.Log.Gen.info($"StreamSerializationUtils.deserialize_column_info bytes: {hex_txt}");
var hex_txt = to_hex(br.BaseStream);
bts.Log.Gen.info($"StreamSerializationUtils.deserialize_column_info bytes: {hex_txt}");
int col_no = br.ReadInt32();
string col_name = br.read_string_cpp();
int tc_int = br.ReadInt32();
@@ -19,6 +19,7 @@ namespace bts {
case StreamIOMessage.StreamOp.CREATE:
// <op, schema, n_rows, rows>
_schema = StreamSerializationUtils.deserialize_schema(ref br);
Log.notnull(_schema);
_row_deserializer = new UniversalRowDeserializer(_schema);
var rows = _get_length_prefix_rows(ref br);
StreamIOMessage rv = new StreamIOMessage(new StreamIOMessage.CreateMessage(rows));
@@ -47,4 +48,4 @@ namespace bts {
private bool _received_create => _schema != null;
}
}
}
@@ -1,6 +1,8 @@
namespace bts {
using BCG;
using Bts.PubSub;
using System;
using System.Collections.Generic;
using System.IO;
@@ -10,11 +12,13 @@ namespace bts {
//byte[] _state = null;
private MemoryStream _str = new MemoryStream();
private BinaryWriter _bw;
public PubsubServerGenericDataHandler() {
_bw = new BinaryWriter(_str);
}
public string TagType => "*";
public string Tag => ".*";
public string TagType => ".*";
public void update(string tag, byte[] msg) {
_str.Position = 0;
@@ -28,6 +32,8 @@ namespace bts {
Log.Gen.info($"PubsubServerGenericDataHandler.get_content tag: {tag} rv: {rv.Length}");
return rv;
}
public void register(InternalPublishDelegate callback) { }
public void doWork() { }
}
// This class is created by the SingleTagData for stream tag_type. Every pub
@@ -40,24 +46,31 @@ namespace bts {
// this is where the state of the table is kept.
BlueTable _state = null;
public PubsubServerSingleStreamTagHandler(string tag) {
Tag = tag;
}
public string TagType => DSTagManager.DataStreamTagType;
public string Tag { get; }
public void update(string tag, byte[] msg) {
DSTagManager.ValidateStreamTag(tag);
StreamIOMessage str_msg = _deser.deserialize(msg);
var state_txt = _state == null ? "NULL" : _state.Count.ToString();
Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (pre.1) tag: {tag} sz: {msg.Length} _state: {state_txt}");
Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (pre.1) tag: {tag} sz: {msg.Length} str_msg: {str_msg} _state: {state_txt}");
switch (str_msg.op) {
case StreamIOMessage.StreamOp.CREATE:
//Log.assert(_state == null);
Log.isnull(_state);
_state = new BlueTable(str_msg.create_message.rows);
break;
case StreamIOMessage.StreamOp.MULTIPLE_REMOVE_ADD:
Log.notnull(_state);
foreach (IBlueRow r in str_msg.multiple_remove_add_message.removed_rows)
_state.deleteRow(r);
_state.addRows(str_msg.multiple_remove_add_message.added_rows);
break;
}
Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (post) tag: {tag} sz: {msg.Length} state: {_state.Count}");
Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (post) tag: {tag} sz: {msg.Length} str_msg: {str_msg} state: {_state.Count}");
}
public byte[] get_content(string tag) {
@@ -75,17 +88,18 @@ namespace bts {
Log.Gen.info($"PubsubServerSingleStreamTagHandler.get_content rv: {rv_txt}");
return rv;
}
public void register(InternalPublishDelegate callback) {
}
public void doWork() { }
}
public class PubsubServerStreamTagHandler : IPubSubServerTagHandler {
IDictionary<string, PubsubServerSingleStreamTagHandler> _handler_map
= new Dictionary<string, PubsubServerSingleStreamTagHandler>();
private readonly string _tag;
public PubsubServerStreamTagHandler() {
}
public string TagType => DSTagManager.DataStreamTagType;
public string Tag => ".*";
public void update(string tag, byte[] msg) {
BCG.IPubSubServerTagHandler rv = _get_handler(tag);
@@ -97,12 +111,21 @@ namespace bts {
}
private BCG.IPubSubServerTagHandler _get_handler(string tag) {
DSTagManager.ValidateStreamTag(tag);
PubsubServerSingleStreamTagHandler rv = null;
if (!_handler_map.TryGetValue(tag, out rv)) {
rv = new PubsubServerSingleStreamTagHandler();
rv = new PubsubServerSingleStreamTagHandler(tag);
_handler_map.Add(tag, rv);
}
return rv;
}
public void register(InternalPublishDelegate callback) {
foreach (var h in _handler_map.Values)
h.register(callback);
}
public void doWork() {
foreach (var h in _handler_map.Values)
h.doWork();
}
}
}
@@ -155,12 +155,12 @@ namespace BCG
"Null arguments are not supported.");
myTag = myGetTag(ci, args, db != null);
mySetRealSource(new TableStream("prepub_" + myTag, db));
mySetRealSource(AttachStream(new TableStream("prepub_" + myTag)));
}
catch (MyException e)
{
// BlueEventLog2.instance().logEvent(e.Message, BlueEventLog2Level.Fatal);
mySetRealSource(new TableStream("error_tag"));
mySetRealSource(AttachStream(new TableStream("error_tag")));
}
}
@@ -9,13 +9,16 @@ using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
public static class DataStreamCoreServices {
public static IServiceCollection AddViewStreamDatabase(this IServiceCollection services, IConfiguration configurationSection) =>
services
.Configure<ViewStreamDatabaseConfig>(configurationSection)
.AddSubscriberCore()
.AddSingleton<ViewStreamDatabase>()
.AddSingleton<IWorker, ViewStreamDatabase>(sp => sp.GetRequiredService<ViewStreamDatabase>());
public static IServiceCollection AddPubTableBroker(this IServiceCollection services) =>
services.AddSingleton<IPubTableBroker, PubTableBroker>();
public static IServiceCollection AddViewStreamDatabase(this IServiceCollection services, IConfiguration configurationSection) =>
services.Configure<WorkerServiceConfig>(configurationSection)
.AddSingleton<IWorker, ViewStreamDatabase>(sp => {
ViewStreamDatabase.SetGlobalInstance(new ViewStreamDatabase(sp.GetRequiredService<ISubscriberCore>()));
return ViewStreamDatabase.instance;
});
public static IServiceCollection AddPrimitiveManager(this IServiceCollection services) =>
services.AddSingleton<PrimitiveManager>();
}
@@ -0,0 +1,34 @@
// namespace Bts.DataStreamCore.Services;
// using BCG;
// using bts;
// using Microsoft.Extensions.Hosting;
// using System.Threading;
// using System.Threading.Tasks;
// public class ViewStreamDatabaseService : IHostedService {
// readonly ViewStreamDatabase viewStreamDatabase;
// readonly MasterWorker masterWorker;
// readonly ViewStreamDatabaseServiceConfig config;
// public ViewStreamDatabaseService(ViewStreamDatabase viewStreamDatabase, MasterWorker masterWorker, ViewStreamDatabaseServiceConfig config) {
// this.viewStreamDatabase = viewStreamDatabase;
// this.masterWorker = masterWorker;
// this.config = config;
// }
// public Task StartAsync(CancellationToken cancellationToken) {
// Log.Gen.info($"Adding viewStreamDatabase to MasterWorker with. Interval: {config.Interval}");
// masterWorker.addWorker(viewStreamDatabase, config.Interval);
// return Task.CompletedTask;
// }
// public Task StopAsync(CancellationToken cancellationToken) {
// Log.Gen.info($"Removing viewStreamDatabase from MasterWorker. Interval: {config.Interval}");
// masterWorker.removeWorker(viewStreamDatabase);
// return Task.CompletedTask;
// }
// }
@@ -0,0 +1,10 @@
namespace Bts.DataStreamCore;
using System;
public class ViewStreamDatabaseConfig {
/// <summary>
/// The interval between 'doWork' calls (i.e., sleep time in loop)
/// </summary>
public TimeSpan Interval { get; set; }
}
@@ -1,39 +1,35 @@
#nullable enable
namespace BCG;
using bts;
namespace BCG {
// this is the stream that gets tables directly off the pubsubServer
public class TableStream : ViewStream {
private readonly string _table_name;
private readonly ITableAccumulator _ta;
using Bts.PubSub;
// this is the stream that gets tables directly off the pubsubServer
public class TableStream : ViewStream {
readonly string _table_name;
ITableAccumulator? _ta;
public TableStream(string table_name)
:
this(table_name,
ViewStreamDatabase.instance) { }
public TableStream(string table_name) =>
_table_name = table_name;
public override string ToString() =>
$"{GetType()} table: {_table_name}";
public TableStream(string table_name, ViewStreamDatabase db) {
_table_name = table_name;
db = db == null ? ViewStreamDatabase.instance : db;
_ta = db.getITableAccumulator(_table_name);
}
protected override void myOnFirstSubscriber() =>
Log.notnull(_ta).on_update += _handle_updates;
protected override void myOnFirstSubscriber() {
_ta.on_update += new TableUpdateDelegate(_handle_updates);
}
protected override void myOnLastSubscriber() =>
Log.notnull(_ta).on_update -= _handle_updates;
protected override void myOnLastSubscriber() {
_ta.on_update -= new TableUpdateDelegate(_handle_updates);
}
protected virtual void _handle_updates(
IBlueRowCollection new_rows, IBlueRowCollection added_rows, IBlueRowCollection del_rows) {
myFireEvent(new DataPacket(new_rows, added_rows, del_rows));
}
protected virtual void _handle_updates(
IBlueRowCollection new_rows, IBlueRowCollection added_rows, IBlueRowCollection del_rows) {
myFireEvent(new DataPacket(new_rows, added_rows, del_rows));
}
public override string ToString() {
return $"{GetType()} table: {_table_name}";
}
protected override void OnViewStreamDatabase(IWorkScheduler viewStreamDatabase) {
_ta = (viewStreamDatabase as ViewStreamDatabase)?.getITableAccumulator(_table_name);
Log.notnull(_ta, "ITableAccumulator is null");
base.OnViewStreamDatabase(viewStreamDatabase);
}
}
@@ -1,11 +1,21 @@
using System;
using BCG.Utilities;
using BCG.Utilities.General;
using bts;
using Bts.DataStream;
using Bts.DataStreamCore;
using Bts.Extensions.DependencyInjection;
using Bts.PubSub;
using mbts.config;
using Microsoft.Extensions.DependencyInjection;
using Spark.Core.Disposables;
namespace BCG
{
/// <summary>
@@ -14,6 +24,10 @@ namespace BCG
public abstract class ViewStream : DataStream
{
private IBlueRowCollection myLastValueForTestMode;
private IServiceProvider serviceProvider;
// private IWorkScheduler workScheduler;
private IWorkScheduler viewStreamDatabase;
IDisposable worker;
protected bool myUseUncopiedNewRowsInFireEvent =
Configuration.get<DataStreamConfig>().UseUncopiedNewRowsInFire;
@@ -25,6 +39,11 @@ namespace BCG
public ViewStream()
{}
#nullable enable
event Action? SubscriptionStarted;
event Action? SubscriptionStopped;
#nullable disable
public IBlueRowCollection rows
{
get
@@ -56,6 +75,23 @@ namespace BCG
}
}
protected IServiceProvider ServiceProvider => serviceProvider;
// protected IWorkScheduler WorkScheduler => workScheduler ??= serviceProvider?.GetService<IWorkScheduler>();
protected IWorkScheduler ViewStreamDatabase => viewStreamDatabase ??= serviceProvider?.GetService<ViewStreamDatabase>();
protected event Action<IServiceProvider> ServiceProviderChanged;
protected event Action<IWorkScheduler> ViewStreamDatabaseChanged;
protected override void myOnFirstSubscriber() {
base.myOnFirstSubscriber();
SubscriptionStarted?.Invoke();
}
protected override void myOnLastSubscriber() {
base.myOnLastSubscriber();
SubscriptionStopped?.Invoke();
}
protected override object myGetValueForInitialSubscription()
{
if (myLastValue == null)
@@ -95,6 +131,53 @@ namespace BCG
}
public void Reset(IServiceProvider serviceProvider) {
Disposable.dispose(ref worker);
Log.isnull(this.serviceProvider);
Log.isnull(ViewStreamDatabase);
this.serviceProvider = serviceProvider;
OnServiceProvider(serviceProvider);
ServiceProviderChanged?.Invoke(ServiceProvider);
// if (workScheduler is not null) {
// OnWorkScheduler(workScheduler);
// }
if (ViewStreamDatabase is not null) {
OnViewStreamDatabase(ViewStreamDatabase);
ViewStreamDatabaseChanged?.Invoke(ViewStreamDatabase);
}
}
public void Dispose() {
Disposable.dispose(ref worker);
}
protected virtual void OnServiceProvider(IServiceProvider serviceProvider) { }
protected virtual void OnViewStreamDatabase(IWorkScheduler viewStreamDatabase) {
var w = this as IWorker;
if (w is not null) {
worker = viewStreamDatabase.Subscribe(w);
}
}
// protected virtual void OnWorkScheduler(IWorkScheduler workScheduler) { }
/// <summary>
/// Attaches <paramref name="stream"/> to the ServiceProvider in this instance. It is designed
/// for cases where a <see cref="ViewStream"/> instantiates other streams (commonly done in
/// the constructor before calling <see cref="DataStream.mySetRealSource(DataStream)"/>).
/// </summary>
/// <typeparam name="T">The type of <see cref="ViewStream"/></typeparam>
/// <param name="stream">The stream to attach.</param>
/// <returns></returns>
protected T AttachStream<T>(T stream) where T : ViewStream {
ServiceProviderChanged += stream.Reset;
if (ServiceProvider is not null) {
stream.Reset(ServiceProvider);
}
return stream;
}
private void myDummy (object o)
{}
}
@@ -9,16 +9,14 @@ using System.Threading;
using mbts;
using System.Linq;
using System.Diagnostics;
using bts;
using Bts.DataStreamCore;
using Bts.PubSub;
using Microsoft.Extensions.Options;
namespace BCG {
public class ViewStreamDatabase : IWorker, IDisposable {
public class ViewStreamDatabase : IViewStreamDatabase, IDisposable {
internal delegate void TagFireDelegate(string tag);
public delegate void DoWorkDelegate();
private event DoWorkDelegate onDoWorkStart,
onDoWorkEnd;
private event TagFireDelegate onTagFireStart,
onTagFireEnd;
//currently onTagUpdate is usd by the DataStreamPerfRecorder to keep track of how often the updates for various streams come in
//and how big they are.
//public delegate void OnTagUpdateDelegate(PubMetaData metaData);
@@ -26,6 +24,7 @@ namespace BCG {
private Dictionary<string, ITableAccumulator> _table_accumulator_map = new Dictionary<string, ITableAccumulator>();
private static ViewStreamDatabase mySelf;
private ISubscriberCore mySubscriberCore;
readonly ViewStreamDatabaseConfig config;
private List<IWorker> _workers = new List<IWorker>();
private bool myIsDisposed = false;
private bool _is_in_do_work = false;
@@ -38,50 +37,16 @@ namespace BCG {
private static readonly object myInstanceLockObject = new object();
private readonly object myStateLock = new object();
public ViewStreamDatabase()
: this(new SubscriberCore()) {
}
public ViewStreamDatabase(ISubscriberCore subscriberCore, IOptions<ViewStreamDatabaseConfig> config)
: this(subscriberCore, config.Value) { }
public ViewStreamDatabase(ISubscriberCore subscriberCore) {
public ViewStreamDatabase(ISubscriberCore subscriberCore, ViewStreamDatabaseConfig config) {
mySubscriberCore = subscriberCore;
this.config = config;
// if (DataStreamLoadFileConfig.instance.profileStreams)
// new DataStreamPerfRecorder(this);
}
private static int myGetTimeInterval() {
if (myTimeInterval == null) {
// TODO
string timeInterval = String.Empty; //ConfigurationRequester.request("viewStreamDatabaseTimeOut");
if (timeInterval == null || timeInterval == "")
myTimeInterval = 0;
else
myTimeInterval = Int32.Parse(timeInterval);
}
return (int)myTimeInterval;
}
public static int instance_hashcode = -1;
public static ViewStreamDatabase instance {
get {
if (mySelf == null || mySelf.myIsDisposed)
lock (myInstanceLockObject) {
if (mySelf == null || mySelf.myIsDisposed) {
mySelf = new ViewStreamDatabase();
instance_hashcode = mySelf.GetHashCode();
}
}
return mySelf;
}
set {
lock (myInstanceLockObject) {
if (mySelf != null)
mySelf.mySubscriberCore.Dispose();
mySelf = value;
instance_hashcode = mySelf.GetHashCode();
}
}
}
public ISubscriberCore subscriber { get { return mySubscriberCore; } }
@@ -108,9 +73,6 @@ namespace BCG {
else {
_is_worker_call_in_progress = true; // to allow reentrancy
try {
if (onDoWorkStart != null)
onDoWorkStart();
var updates = mySubscriberCore.get_updates(TimeSpan.FromMilliseconds(10));
if (updates != null) {
var consolidated_update_map = _consolidate_updates(updates);
@@ -125,11 +87,7 @@ namespace BCG {
}
_timer.Restart();
if (onTagFireStart != null)
onTagFireStart(tag);
table_accumulator.handle_updates(tag_updates);
if (onTagFireEnd != null)
onTagFireEnd(tag);
_timer.Stop();
if (_timer.Elapsed > myLogTimeThreshold) {
bts.Log.Gen.error($"ViewStreamDatabase.doWork tag: {tag} took {_timer.Elapsed}");
@@ -160,8 +118,6 @@ namespace BCG {
_pending_worker_ops.Clear();
_is_in_do_work = false;
_is_worker_call_in_progress = false;
if (onDoWorkEnd != null)
onDoWorkEnd();
}
}
}
@@ -222,6 +178,12 @@ namespace BCG {
_workers.Remove(w);
}
void IWorkScheduler.Add(IWorker worker, TimeSpan interval) =>
addWorker(worker);
void IWorkScheduler.Remove(IWorker worker) =>
removeWorker(worker);
#region IDisposable Members
public void Dispose() {
@@ -231,11 +193,6 @@ namespace BCG {
}
}
internal static void SetGlobalInstance(ViewStreamDatabase vsd) {
Log.isnull(mySelf);
mySelf = vsd;
}
private List<PendingAddRemoveWorker> _pending_worker_ops = new List<PendingAddRemoveWorker>();
private bool _is_worker_call_in_progress = false;
class PendingAddRemoveWorker {
@@ -11,16 +11,16 @@ class Assignment : StatementExpression {
expression = st;
}
public override string print(int startCol, IList<string> activeBindings) {
public override string print(int startCol, IList<string> activeBindings, PrimitiveManager primitiveManager) {
string prelude = id + " = ";
string result = prelude + expression.print(startCol + prelude.Length, activeBindings);
string result = prelude + expression.print(startCol + prelude.Length, activeBindings, primitiveManager);
if (activeBindings.IndexOf(id) == -1)
activeBindings.Add(id);
return result;
}
public override ViewStream? myGetView(IDictionary<string, StatementExpression> env) {
var result = expression.getView(env);
public override ViewStream? myGetView(IDictionary<string, StatementExpression> env, PrimitiveManager primitiveManager) {
var result = expression.getView(env, primitiveManager);
env[id] = expression;
return result;
}
@@ -10,11 +10,11 @@ public class DataStreamExpression {
public ImmutableList<StatementExpression> Statements { get; }
public string print(int startPos, IList<string> activeBindings) {
public string print(int startPos, IList<string> activeBindings, PrimitiveManager primitiveManager) {
var sb = new StringBuilder();
foreach (var st in Statements) {
sb.Append(new string(' ', startPos));
sb.Append(st.print(startPos, activeBindings));
sb.Append(st.print(startPos, activeBindings, primitiveManager));
sb.AppendLine(";");
}
sb.AppendLine();
@@ -31,10 +31,10 @@ public class DataStreamExpression {
return sb.ToString();
}
public ViewStream? getView(IDictionary<string, StatementExpression> env) {
public ViewStream? getView(IDictionary<string, StatementExpression> env, PrimitiveManager primitiveManager) {
for (int i = 0; i < Statements.Count - 1; i++) {
Statements[i].getView(env);
Statements[i].getView(env, primitiveManager);
}
return Statements[^1].getView(env);
return Statements[^1].getView(env, primitiveManager);
}
}
Loaded 100 of 208 files, more files were not shown because too many files have changed in this diff. Show more