Compare commits

..
Author SHA1 Message Date
John Colvin ba664b95b8 fix use dep injection in bts.datastreamwebapp.server 2024-07-11 11:06:29 -04:00
John Colvin d6f370f862 fix Add spark service modules designed for DependencyInjection 2024-07-11 11:05:23 -04:00
John Colvin 079826f4de Fix mismatched types in PubsubHeartbeatStream schema 2024-07-11 10:59:41 -04:00
John Colvin 2eee3cdaab Change DataStreamPublisher json fields to match cs class property names verbatim 2024-07-11 10:43:24 -04:00
John Colvin 5e3215ee52 fix: Always call 'handle_internal_pubs' in PubSubMessageReactor main loop 2024-07-11 10:38:30 -04:00
John Colvin 7739cc2f3d Use Dependency Injection in Bts.DataStreamWebApp.Server 2024-07-11 10:37:29 -04:00
John Colvin 8347a7c126 refactor: Change Bts.DataStreamWebApp.Server framework from net8 -> net7
Although it would build on net8, all other cs code targets net7.  Furthermore,
VS Code intellisense seems less reliable/crashes frequently when the sln file
targets both frameworks).
2024-07-11 10:37:28 -04:00
John Colvin fbfc960df6 improvement: Make DataStreamSubscription an IHostedService
To properly manage its life cycle
2024-07-11 10:37:28 -04:00
John Colvin 451f721030 improvement: Accept CancellationToken, catch exceptions when running ds websocket server 2024-07-11 10:37:28 -04:00
John Colvin bb992c0410 Fix mismatched types in PubSubServerTagStatsStream schema 2024-07-11 10:37:28 -04:00
John Colvin db18aebf92 Add DataStreamPublisher service modules designed for Dependency Injection 2024-07-11 10:37:28 -04:00
John Colvin 8127c9c9fe refactor: Decouple ViewPublisher from IPubTableBroker impl 2024-07-11 10:37:28 -04:00
John Colvin 1825b339b1 Change PubSubServer json fields to match cs class property names verbatim 2024-07-11 10:37:28 -04:00
John Colvin 03ab3fbe90 Get rid of JsonPropertyName attributes in PubSubConfig
The Dependency Injection framework (based around dotnet ApplicationBuilder)
does not support this attribute, and maps property names verbatim to json fields.
2024-07-11 10:37:28 -04:00
John Colvin b61bf16a42 Add PubSub service modules designed for Dependency Injection 2024-07-11 10:37:28 -04:00
John Colvin d969489bf4 refactor: Decouple RemoteQueryManager from IManagedSubscriber, IPubTableBroker impl 2024-07-11 10:37:28 -04:00
John Colvin 9cd4974f49 refactor: Decouple ManagedSubscriber from ISubscriberCore impl 2024-07-11 10:37:28 -04:00
John Colvin 793634c8cb refactor: Only support 1 tag type in IPubSubServerTagHandler 2024-07-11 10:37:28 -04:00
John Colvin eae20b6c94 refactor: Decouple PubTableBroker from IPublisherCore impl
Reference interface rather than the PublisherCore impl.
2024-07-11 10:37:28 -04:00
John Colvin 2848bb6a5d improvement: Accept CancellationToken in JsonRpcRequestObservable 2024-07-11 10:37:16 -04:00
John Colvin 687324998b Add spark service modules designed for Dependency Injection 2024-07-11 10:36:04 -04:00
John Colvin 97cebd62d4 fix: Dispose market data sub in various md streams 2024-07-11 10:36:04 -04:00
John Colvin 492ca0c28f cleanup: Remove unused pubsub tag handler code
* Removed 'IPubSubServerTagHandler.register'
  - All implementations of this function were empty
* Removed 'IPubSubServerTagHandler.supportedTagTypes'
  - This property wasn't referenced anywhere
* Remove IWorker inheritance from IPubSubServerTagHandler
  - No reason these interfaces have to be tied together
2024-07-10 13:35:18 -04:00
John Colvin baa5019039 refactor: Rename CloudStreamManager.cs -> RemoteQueryManager.cs
This matches the file name
2024-07-10 01:18:21 -04:00
John Colvin ac2d31e1fc 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-07-09 15:16:36 -04:00
John Colvin 9821134bc3 Change pub sub schema prefix 'bluto://' -> 'bts-cloud://'
In config json files (not actual code change)
2024-07-09 15:14:30 -04:00
John Colvin 9cf892322d fix build: Fix prop names after refactoring PubSubRoute
* Change 'server_address' -> 'ServerAddress'
* Change 'tag_reg_exp' -> 'TagRegExp'
2024-07-09 15:14:30 -04:00
John Colvin 77bf91be1e build: Add scripts to publish data stream apps 2024-07-09 15:14:30 -04:00
John Colvin 63b064ac79 Fix typo in log msg 2024-07-09 15:14:30 -04:00
John Colvin b617775c29 Fix net7 -> net8 typo in publish_data_stream_web_app.py
The DataStream Web App is on net8
2024-07-09 15:14:30 -04:00
John Colvin 4f7d77420c 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-09 15:14:30 -04:00
John Colvin 3ba6a91ca7 vscode: Add DataStreamPublisher debug cfg to skip build 2024-07-09 15:14:30 -04:00
John Colvin 0abde9eb7c 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-09 15:14:30 -04:00
John Colvin 1531a57bc5 feat: Add DataStream WebApp
Included in DataStreamSuite.sln, but not DataStreamQueryAnalyzer.sln
2024-07-09 15:14:29 -04:00
John Colvin a62ea64c5e 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-07-09 15:14:29 -04:00
John Colvin 450773f588 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-07-09 15:14:29 -04:00
John Colvin f09d6badaf fix build (api server): Change namespace to bts.websocket_core 2024-07-09 15:14:29 -04:00
John Colvin 6b054e0718 Remove docker submodules again 2024-07-09 15:14:29 -04:00
John Colvin 5e2789d16e 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-07-09 15:14:29 -04:00
John Colvin 9ebdb2e460 cleanup: Fix benign typos in build scripts
Change BUILD_SPARK_PY var name to BUILD_PY (because these scripts to actually build 'spark')
2024-07-09 15:14:29 -04:00
John Colvin d7bf82722b fix build (warn): Only use Registry on windows 2024-07-09 15:14:29 -04:00
John Colvin e1526e25a7 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-07-09 15:14:26 -04:00
John Colvin 089ba9b369 cleanup: Normalize indentation in TradeBarsStream 2024-07-09 15:09:02 -04:00
John Colvin 567976043f cleanup: Disable nullability warnings
* BZip2InputStream
* TarInputStream
* Inflater
* InflaterDynHeader
* BZip2OutputStream
* DeflaterHuffman
* ParallelJoinTest
2024-07-09 15:07:24 -04:00
John Colvin 2fd3382939 cleanup: Modernize PubSubRoute code, json parsing 2024-07-09 15:06:34 -04:00
John Colvin a96f1c8f17 cleanup: Remove unused DateTime grammar from exp parser 2024-07-09 15:03:30 -04:00
John Colvin 6f623f9a37 Add JsonRpcRequestObservable, AsyncEnumerable 2024-07-09 15:03:30 -04:00
John Colvin 827189cdb4 Add json facilities for DataStream websocket api 2024-07-09 15:03:30 -04:00
John Colvin 123f12ab88 refactor: Add handy WebsocketExtensions
* IsSocketStateOpen
* IsClosedProperly
2024-07-09 14:59:06 -04:00
John Colvin f51ba8fb56 feat: Add JsonRpc types for websocket paradigm 2024-07-09 14:59:06 -04:00
John Colvin 2506fb4f2b cleanup: Remove empty file (clearly added by accident) 2024-07-09 14:59:06 -04:00
John Colvin f92964114e 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-07-09 14:59:06 -04:00
John Colvin baa0f1430b 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-07-09 14:59:06 -04:00
John Colvin 7104bd787a refactor: Add WebsocketClient class
Basically an ISubject (I/O) wrapper for IWebsocketConnectionGateway
2024-07-09 14:59:06 -04:00
John Colvin 5db92d1f7d refactor: Change WebsocketServerConfig.port to uint32
It should always be a number between 1-65535
2024-07-09 14:59:06 -04:00
John Colvin eb0fcf0b84 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-07-09 14:59:06 -04:00
John Colvin a52b187313 refactor: Remove nullability constraint in GetValueOrDefault 2024-07-09 14:59:06 -04:00
John Colvin 669b059eb2 refactor: Use events to avoid circular deps for ws sync/gateway 2024-07-09 14:59:06 -04:00
John Colvin 36ab8492d3 refactor: Let caller specify nullability of T in json parse fn 2024-07-09 14:48:59 -04:00
John Colvin 54f37b2401 Merge remote-tracking branch 'origin/development' into development.ps_linux 2024-07-09 14:46:18 -04:00
Taha Afzal 5088b5ebe0 Fixing problems where PSS's client socket's IP was different than what the client was sending in PubSubClientSubscription 2024-07-08 23:09:53 +00:00
Taha Afzal 010dd3a26f Fixing PubsubEchoTestStream 2024-07-07 06:13:22 +00:00
Taha Afzal 1a18755176 Added pss_echo() 2024-07-06 06:53:35 +00:00
Taha Afzal 233e2576af Merging 2024-07-05 21:21:27 +00:00
Taha Afzal 543c655bca src/DataStreamSuite/PubSub/ManagedTCP/AsyncTCPConnection_AsyncWrapper.cs 2024-07-05 21:13:36 +00:00
Taha 050e21e22b Adding TableMetaDataStream 2024-07-04 23:39:04 -04:00
Taha Afzal 672ad6d237 Cleaning Config 1.0 (not completely done) 2024-06-30 23:17:20 -04:00
Taha 0a8f6e866f Added pubsub routes to the Title of DSQA 2024-06-27 14:04:07 -04:00
Taha b177ffef04 Fixing the bug where messages were being sent by the pubsubserver out of order 2024-06-27 00:17:22 -04:00
Taha 393f304be4 some buf gixes + cleanup 2024-06-26 00:23:19 -04:00
Taha 151816bf5e PubSuMessageReactor empties the message queue (instead oif one elem ata a time) in the main loop 2024-06-24 02:29:28 -04:00
Taha 87b6a2044c PubSubServer's linux issues fixed 2024-06-23 21:13:37 -04:00
208 changed files with 2782 additions and 3264 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/net8.0/Bts.DataStreamWebApp.Server.dll",
"program": "${workspaceFolder}/build/Debug/bin/net7.0/Bts.DataStreamWebApp.Server.dll",
"args": [
"--sparkd",
"http://localhost:8000/config",
"--config",
"${workspaceFolder}/src/DataStreamSuite/Bts.DataStreamWebApp.Server/Configurations/DevServer.json",
],
"env": {
"ASPNETCORE_ENVIRONMENT": "Development",
@@ -25,11 +25,11 @@
"name": "Bts.DataStreamWebApp.Server (Skip Build)",
"type": "coreclr",
"request": "launch",
"program": "${workspaceFolder}/build/Debug/bin/net8.0/Bts.DataStreamWebApp.Server.dll",
// "args": [
// "--config",
// "${workspaceFolder}/src/DataStreamSuite/Bts.DataStreamWebApp.Server/Configurations/DevServer.json",
// ],
"program": "${workspaceFolder}/build/Debug/bin/net7.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,5 +1,4 @@
using System;
using System.Runtime.InteropServices;
using System.Runtime.Versioning;
using Microsoft.Win32;
@@ -16,8 +15,11 @@ namespace BCG
[SupportedOSPlatform("windows")]
public static double getValue (RegistryKey rk, string name)
{
var str = rk.GetValue(name)?.ToString();
return str is not null && double.TryParse(str, out var dbl) ? dbl : double.NaN;
object v = rk.GetValue (name);
if (v != null)
return double.Parse (v.ToString ());
else
return double.NaN;
}
}
}
@@ -4,7 +4,7 @@
font-weight: 400;
color-scheme: light dark;
color: rgba(0, 0, 0, 0.87);
color: rgba(255, 255, 255, 0.87);
background-color: #242424;
font-synthesis: none;
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<TargetFramework>net7.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="8.0.6" />
<PackageReference Include="Microsoft.AspNetCore.SpaServices.Extensions" Version="8.0.6" />
<PackageReference Include="Microsoft.AspNetCore.SpaProxy" Version="7.0.20" />
<PackageReference Include="Microsoft.AspNetCore.SpaServices.Extensions" Version="7.0.20" />
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
<PackageReference Include="Swashbuckle.AspNetCore" Version="6.4.0" />
</ItemGroup>
@@ -27,7 +27,6 @@
<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,15 +1,36 @@
{
"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": [
{
@@ -25,9 +46,61 @@
"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,36 +1,30 @@
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, PrimitiveManager primitiveManager, IServiceProvider serviceProvider) {
public DataStreamSubscriptionController(ILogger<DataStreamSubscriptionController> logger) {
this.logger = logger;
this.primitiveManager = primitiveManager;
this.serviceProvider = serviceProvider;
logger.LogInformation("Creating {ControllerType}", GetType().Name);
}
[Route("/ws")]
public async Task Get() {
public async Task Get(CancellationToken cancellationToken) {
if (!HttpContext.WebSockets.IsWebSocketRequest) {
logger.LogError("Expected websocket request on path '{}'", HttpContext.Request.Path);
return;
}
logger.LogInformation("Establishing websocket connection...");
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();
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);
}
}
}
@@ -12,22 +12,23 @@ using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
class DataStreamSubscription : IDisposable {
class DataStreamSubscription : IDisposable, IHostedService {
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, ViewStream viewStream, WebSocket webSocket, ILogger logger) {
public DataStreamSubscription(string id, JsonRpcId requestId, 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() {
@@ -40,6 +41,12 @@ class DataStreamSubscription : IDisposable {
ViewStream.onChange += onChange;
}
public Task StartAsync(CancellationToken cancellationToken) {
cancellationToken.ThrowIfCancellationRequested();
Start();
return Task.CompletedTask;
}
public Task Clear() {
if (LastFired is null) {
return Task.CompletedTask;
@@ -58,6 +65,12 @@ class DataStreamSubscription : IDisposable {
ViewStream.onChange -= onChange;
}
public Task StopAsync(CancellationToken cancellationToken) {
cancellationToken.ThrowIfCancellationRequested();
Stop();
return Task.CompletedTask;
}
public void Dispose() =>
Stop();
@@ -1,7 +1,5 @@
namespace Bts.DataStreamWebApp.Server.Controllers;
using BCG;
using bts.websocket_core.json_rpc;
using Bts.DataStreamWebApp.Server.Messages;
@@ -11,26 +9,22 @@ using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
public class DataStreamWebSocketServer {
public class DataStreamWebSocketServer : IDisposable, IAsyncDisposable {
readonly ILogger logger;
readonly WebSocket webSocket;
readonly Dictionary<string, DataStreamSubscription> subscriptions = new();
readonly PrimitiveManager primitiveManager;
readonly IServiceProvider serviceProvider;
public DataStreamWebSocketServer(WebSocket webSocket, ILogger logger, PrimitiveManager primitiveManager, IServiceProvider serviceProvider) {
public DataStreamWebSocketServer(WebSocket webSocket, ILogger logger) {
this.webSocket = webSocket;
this.logger = logger;
this.primitiveManager = primitiveManager;
this.serviceProvider = serviceProvider;
}
public async Task<int> Run() {
public async Task<int> Run(CancellationToken cancellationToken) {
logger.LogInformation("DataStreamWebSocketServer running");
await foreach (var req in JsonRpcRequestAsyncEnumerable.Create(webSocket, logger)) {
await foreach (var req in JsonRpcRequestAsyncEnumerable.Create(webSocket, logger, cancellationToken)) {
await (req.Method switch {
"subscribe" => onSubscribe(req),
"unsubscribe" => onUnsubscribe(req.Deserialize<UnsubscribeRequest>()),
"subscribe" => onSubscribe(req, cancellationToken),
"unsubscribe" => onUnsubscribe(req.Deserialize<UnsubscribeRequest>(), cancellationToken),
_ => Task.CompletedTask,
});
}
@@ -38,28 +32,51 @@ public class DataStreamWebSocketServer {
return 0;
}
async Task onSubscribe(JsonRpcRequest<JsonNode> req) {
/// <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) {
ArgumentNullException.ThrowIfNull(req.Id);
ArgumentException.ThrowIfNullOrEmpty(req.Method);
if (!SubscribeRequest.ParseQuery(req, primitiveManager, out var vs, out var subId, out var errMsg, out var errCode)) {
if (!SubscribeRequest.ParseQuery(req, out var vs, out var subId, out var errMsg, out var errCode)) {
logger.LogError("Invalid 'subscribe' request: {Reason}", errMsg);
await send(new JsonRpcError<JsonNode>() {
var rpcError = new JsonRpcError<JsonNode>() {
Code = (long)errCode,
Message = $"{errCode}: {errMsg}",
});
};
await send(rpcError, cancellationToken);
return;
}
if (subscriptions.TryGetValue(subId, out var oldSub)) {
oldSub.Dispose();
}
vs.Reset(serviceProvider);
var sub = new DataStreamSubscription(subId, vs, webSocket, logger);
var sub = new DataStreamSubscription(subId, req.Id.Value, vs, webSocket, logger);
subscriptions[subId] = sub;
await sendSubscriptionStatus(req.Id.Value, subId, req.Method, SubscriptionStatus.Subscribed);
sub.Start();
await sendSubscriptionStatus(req.Id.Value, subId, req.Method, SubscriptionStatus.Subscribed, cancellationToken);
await sub.StartAsync(cancellationToken);
}
async Task onUnsubscribe(JsonRpcRequest<UnsubscribeRequest> req) {
async Task onUnsubscribe(JsonRpcRequest<UnsubscribeRequest> req, CancellationToken cancellationToken) {
ArgumentNullException.ThrowIfNull(req.Id);
ArgumentException.ThrowIfNullOrEmpty(req.Method);
var subId = req.Params?.SubscriptionId ?? string.Empty;
@@ -70,24 +87,35 @@ public class DataStreamWebSocketServer {
logger.LogInformation("Disposing subscription");
sub.Dispose();
subscriptions.Remove(subId);
await sendSubscriptionStatus(req.Id.Value, subId, req.Method, SubscriptionStatus.Unsubscribed);
await sendSubscriptionStatus(req.Id.Value, subId, req.Method, SubscriptionStatus.Unsubscribed, cancellationToken);
}
Task sendSubscriptionStatus(JsonRpcId reqId, string subId, string method, SubscriptionStatus status) {
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) {
logger.LogInformation("Sending {Method} ack. SubscriptionId = {SubscriptionId}", method, subId);
return send(JsonRpcResponse.Create(reqId, new SubscriptionStatusMessage() {
var response = JsonRpcResponse.Create(reqId, new SubscriptionStatusMessage() {
SubscriptionId = subId,
Status = status,
}));
});
return send(response, cancellationToken);
}
Task send<T>(T message) {
Task send<T>(T message, CancellationToken cancellationToken) {
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.None);
return webSocket.SendAsync(bytes, WebSocketMessageType.Text, true, cancellationToken);
}
void disposeSubscriptions() {
logger.LogInformation("Disposing DataStreamWebSocketServer subscriptions. Count: {Count}", subscriptions.Count);
subscriptions.Values.DisposeAll();
subscriptions.Clear();
}
}
@@ -18,14 +18,7 @@ public class SubscribeRequest {
[JsonPropertyName("subscription_id")]
public string? SubscriptionId { get; set; }
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
) {
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) {
stream = null;
subscriptionId = null;
errorCode = ErrorCode.None;
@@ -57,7 +50,7 @@ public class SubscribeRequest {
}
string? exceptionMsg = null;
try {
stream = exp.getView(new Dictionary<string, StatementExpression>(0), primitiveManager);
stream = exp.getView(new Dictionary<string, StatementExpression>(0));
} catch (Exception ex) {
exceptionMsg = ex.ToString();
}
@@ -1,9 +1,7 @@
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;
@@ -57,8 +55,5 @@ static void ConfigureServices(IServiceCollection services, IConfiguration config
services.AddMasterWorker(config.GetSection("MasterWorker"));
services.AddViewStreamDatabase(config.GetSection("ViewStreamDatabase"));
services.AddPubSubClient(config.GetSection("PubSub"));
services.AddPrimitiveManager();
services.AddSpark(config.GetSection("Spark"));
services.AddSparkd(config.GetSection("SparkdCredentials"));
services.AddHostedService<DataStreamWebAppService>();
services.AddSparkd(config.GetSection("Sparkd"));
}
@@ -1,65 +0,0 @@
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;
}
}
@@ -1,6 +0,0 @@
{
"WebServer": {
"ContentRootPath": "build/Bts.DataStreamWebApp.Client/production",
"WebRootPath": "dist"
}
}
@@ -5,36 +5,5 @@
"Microsoft.AspNetCore": "Warning"
}
},
"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"
}
"AllowedHosts": "*"
}
@@ -1,13 +0,0 @@
<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>
@@ -1,11 +0,0 @@
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);
}
@@ -1,6 +0,0 @@
#nullable enable
namespace Bts.Extensions.DependencyInjection;
public interface IDependOnKeyed<T> : IDependOn<T> {
string Key { get; }
}
@@ -1,130 +0,0 @@
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,7 +14,6 @@
<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>
@@ -1,35 +0,0 @@
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;
}
}
@@ -1,7 +0,0 @@
namespace Bts.DataStream;
using BCG;
public interface IViewStreamFactory {
public ViewStream? Create();
}
@@ -11,8 +11,11 @@ public class PandasScriptStream : ViewStream {
public PandasScriptStream(ViewStream vs, string file)
: this(vs, file, null) { }
public PandasScriptStream(ViewStream vs, string file, string? pythonExe) {
mySetRealSource(AttachStream<PandasStream>(new(vs, readFile(), pythonExe)));
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));
string readFile() {
try {
@@ -1,6 +1,7 @@
namespace Bts.DataStream;
using BCG;
using BCG.Extensions;
using BCG.Utilities.General;
using bts;
@@ -22,6 +23,7 @@ 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;
@@ -32,11 +34,12 @@ public class PandasStream : ViewStream, IWorker {
IDisposable? dispose;
public PandasStream(ViewStream vs, string dataFrameExpression)
: this(vs, dataFrameExpression, null) { }
: this(vs, dataFrameExpression, null, ViewStreamDatabase.instance) { }
public PandasStream(ViewStream vs, string dataFrameExpression, string? pythonExe) {
public PandasStream(ViewStream vs, string dataFrameExpression, string? pythonExe, ViewStreamDatabase vsd) {
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);
@@ -56,7 +59,7 @@ import sys
}
protected override void myOnFirstSubscriber() {
base.myOnFirstSubscriber();
vsd.addWorker(this);
inputCsvStream.onChange += onInputChanged;
this.dispose = Disposable.Create(dispose);
@@ -65,6 +68,7 @@ import sys
void dispose() {
inputCsvStream.onChange -= onInputChanged;
vsd.removeWorker(this);
deleteFile(inputFile);
deleteFile(outputFile);
}
@@ -81,10 +85,8 @@ import sys
}
}
protected override void myOnLastSubscriber() {
base.myOnLastSubscriber();
protected override void myOnLastSubscriber() =>
Disposable.dispose(ref dispose);
}
static void executePythonCode(string pythonCode, string? pythonExe = null) {
pythonExe ??= "python";
@@ -1,19 +1,16 @@
using BCG.Utilities.General;
using Bts.PubSub;
using Microsoft.Extensions.DependencyInjection;
using System;
namespace BCG {
public class TCPConnectionTestStream : ViewStream, IParameterizedWorker, ITCPConnectionHandler, ITCPMessageHandler {
public class TCPConnectionTestStream : ViewStream, IWorker, 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;
@@ -21,12 +18,14 @@ namespace BCG {
private TimeSpan _send_period;
private ITCPConnection _connection = null;
private byte[] _buff;
WorkActivator workActivator;
public TCPConnectionTestStream(string ip, int port, int msg_size, int send_period_seconds) {
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) {
_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),
@@ -36,21 +35,16 @@ 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() {
workActivator.Start();
MasterWorker.getInstance(false).addWorker(this, _send_period);
var server_rci = new RemoteConnectionInfo(_ip, _port);
bts.AsyncTCPConnector_APM.connect_to(server_rci, this);
}
protected override void myOnLastSubscriber() {
workActivator.Stop();
MasterWorker.getInstance(false).removeWorker(this);
}
public void doWork() {
@@ -66,20 +60,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,14 +2,15 @@ using System;
using System.Collections.Generic;
using Bts.DataStreamExpressions;
using BCG.Utilities.General;
using bts;
using Bts.DataStreamCore;
using Bts.PubSub;
using Bts.DataStreamCore;
namespace BCG {
// listens to the requests posted on "cloud_streams" request table, and
// if unfulfilled starts a ViewPubsliher
public class RemoteQueryManager : IParameterizedWorker {
public class RemoteQueryManager : IWorker {
public class StreamPubUnit {
public string name;
public string stream_expression;
@@ -25,13 +26,12 @@ namespace BCG {
view_publisher = view_publisher_;
}
}
static ViewStream? build_stream(string stream_expression, PrimitiveManager primitiveManager, IServiceProvider serviceProvider)
static ViewStream? build_stream(string stream_expression)
{
try
{
var exp = DataStreamExpressionsInterpreter.Evaluate(stream_expression);
var rv = exp.getView(new Dictionary<string, StatementExpression>(0), primitiveManager);
rv?.Reset(serviceProvider);
var rv = exp.getView(new Dictionary<string, StatementExpression>(0));
return rv;
} catch (Exception ex) {
@@ -43,24 +43,18 @@ namespace BCG {
_stream_pub_map = new Dictionary<string, StreamPubUnit>();
private ViewStream _stream = new TableStream("cloud_streams");
readonly IPubTableBroker _broker;
private readonly ManagedSubscriber _sub;
readonly PrimitiveManager primitiveManager;
readonly IServiceProvider serviceProvider;
readonly IManagedSubscriber _sub;
public RemoteQueryManager(ManagedSubscriber sub, IPubTableBroker broker, PrimitiveManager primitiveManager, IServiceProvider serviceProvider) {
_sub = sub;
_broker = broker;
this.primitiveManager = primitiveManager;
this.serviceProvider = serviceProvider;
public RemoteQueryManager(IManagedSubscriber managedSubscriber, IPubTableBroker pubTableBroker) {
_sub = managedSubscriber;
_broker = pubTableBroker;
_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) {
@@ -79,11 +73,7 @@ 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}");
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;
}
ViewStream stream = build_stream(stream_expression);
_expose_to_ws(name, stream);
if (persist) {
stream = new PersistorStream(stream, name, "C:\\Temp\\", 1000000); // 10MBs
@@ -1,6 +1,6 @@
using bts;
using BCG;
using Microsoft.Extensions.DependencyInjection;
using bts;
using System;
@@ -9,7 +9,7 @@ namespace BCG {
[DataStreamUserRespresentation("remoteQuery")]
public class RemoteQueryStream : ViewStream {
private readonly string _name, _stream_exp;
IPublisherCore? _publisher_core_backing;
private PublisherCore _publisher_core = new PublisherCore();
private ViewStream _source_str;
public RemoteQueryStream(params string[] lines) {
@@ -20,14 +20,7 @@ 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 = AttachStream(new MapStream(vs, _get_schema, _map));
ViewStream rs = new MapStream(vs, (IBlueTableSchema s) => _get_schema(s), (IBlueRow r) => _map(r));
mySetRealSource(rs);
}
@@ -91,14 +91,10 @@ namespace BCG
}
_increment_values = increment_values_del;
_get_column_prefix = get_column_prefix_del;
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);
}
if(only_fire_once)
mySetRealSource(new NonIncGroupByStream(new OnlyFireOnceStream(stream), this));
else
mySetRealSource(new NonIncGroupByStream(stream, this));
}
private void _set_schema(List<string> tc_prefixes, Pair<string, Type>[] group_by_col_info, string table_name)
@@ -245,4 +241,4 @@ namespace BCG
return values;
}
}
}
}
@@ -14,7 +14,7 @@ namespace BCG
public CaseChangeStream(ViewStream input, bool toUpper, params string[] columnsToChangeCase)
{
myCaseChangingColumns = columnsToChangeCase;
mySetRealSource(AttachStream(new SuperMapStream(input, this)));
mySetRealSource(new SuperMapStream(input, this));
myIsUpperCase = toUpper;
}
@@ -32,10 +32,10 @@ namespace BCG
{
mySetRealSource (
AttachStream(new MapStream (sourceStream,
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(AttachStream(new CrushingCollectorStream(stream, this)));
mySetRealSource(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 (AttachStream(new GroupByStream (stream,
new DistinctRowEvaluator (groupByCols.ToArray (typeof(string)) as string[]), false)));
mySetRealSource (new GroupByStream (stream,
new DistinctRowEvaluator (groupByCols.ToArray (typeof(string)) as string[]), false));
}
}
}
@@ -3,20 +3,25 @@ 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,12 +1,7 @@
using BCG.Utilities.General;
using Bts.DataStream;
using Bts.PubSub;
using mbts.config;
using Spark.Core.Disposables;
using System;
using System.Text.RegularExpressions;
@@ -146,7 +141,6 @@ 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;
@@ -174,7 +168,7 @@ namespace BCG
if (myUseNewFilter)
{
myFilterStream = new BCG.StreamWorld.Addition.FilterStream(vs, filter);
mySetRealSource(AttachStream(myFilterStream));
mySetRealSource(myFilterStream);
}
else
{
@@ -266,7 +260,6 @@ namespace BCG
}
else
{
Disposable.dispose(ref filterWorkerSubscription);
myFilter = value;
if (numberOfSubscribers > 0)
{
@@ -281,9 +274,6 @@ 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);
}
}
}
}
@@ -304,14 +294,6 @@ 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,13 +4,11 @@ using System.Linq;
using System.Text;
using BCG.Utilities.General;
using Bts.PubSub;
namespace BCG
{
public class SlidingWindowFilter: IDynamicStreamFilter, IWorker
{
private IWorkScheduler workScheduler;
private ViewStreamDatabase myVSD;
private string myDateColumn;
private int myNSecs;
private DateTime myReference;
@@ -19,9 +17,14 @@ 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);
}
@@ -52,9 +55,5 @@ 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 {
public class TimeBasedFilter : IDynamicStreamFilter, IWorker, IDisposable {
public static TimeSpan parse_duration(string input) {
int days = 0, hours = 0, minutes = 0, seconds = 0;
@@ -42,15 +42,28 @@ namespace BCG {
: this(col_name, parse_duration(row_time_to_live)) {
}
public TimeBasedFilter(string rowTimeColumnName, TimeSpan rowTimeToLive)
: this(rowTimeColumnName, rowTimeToLive, new TimeSpan(0, 1, 0) /* 1 minute */) {
: this(ViewStreamDatabase.instance, rowTimeColumnName, rowTimeToLive, new TimeSpan(0, 1, 0) /* 1 minute */) {
}
public TimeBasedFilter(string rowTimeColumnName, TimeSpan rowTimeToLive, TimeSpan filterFrequency) {
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;
_time_col = rowTimeColumnName;
_ttl = rowTimeToLive;
myFilterFrequency = filterFrequency;
myNextFilter = DateTime.MinValue;
// add worker
myVsd.addWorker(this);
}
#endregion
@@ -84,11 +97,29 @@ 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));
}
}
}
@@ -1,14 +0,0 @@
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(AttachStream(new GroupByStream(stream,
new MaxGroupByEvaluator(outputCol, inputCols))));
mySetRealSource(new GroupByStream(stream,
new MaxGroupByEvaluator(outputCol, inputCols)));
}
public MaxStream(ViewStream stream, string outputCol)
@@ -1,15 +1,13 @@
namespace BCG {
public class MinStream : ViewStream {
public MinStream(ViewStream stream, string outputCol, string[] inputCol) {
mySetRealSource(AttachStream(new GroupByStream(stream,
new MinGroupByEvaluator(outputCol, inputCol))));
}
using System;
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);
}
}
namespace BCG {
public class MinStream : ViewStream {
public MinStream(ViewStream stream, string outputCol, string[] inputCol) {
mySetRealSource(new GroupByStream(stream,
new MinGroupByEvaluator(outputCol, inputCol)));
}
public MinStream(ViewStream stream, string outputCol)
: this(new AddColumnsStream(stream, "fakeCol=0"), outputCol, new string[] { "fakeCol" }) { }
}
}
@@ -26,22 +26,22 @@ namespace BCG {
public SumStream(ViewStream stream, string spaceSeparatedOutputCols, string spaceSeparatedGroupByCols) {
string[] outputCols = spaceSeparatedOutputCols.Split(' ');
string[] groupByCols = spaceSeparatedGroupByCols.Split(' ');
mySetRealSource(AttachStream(new GroupByStream(stream,
mySetRealSource(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 = 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"));
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");
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(AttachStream(new SumStream(stream, output_columns, groupby_columns, force_null_into_other_columns)));
mySetRealSource(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,13 +52,12 @@ namespace BCG {
// rs = new GroupByStream(rs, new MultiColumnSumGroupByEvaluator(output_columns, new string[] { "fake_col" }));
// rs = new ProjectStream(rs, "!fake_col");
} else {
rs = AttachStream(new GroupByStream(stream, new MultiColumnSumGroupByEvaluator(output_columns, groupby_columns)));
rs = 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 = AttachStream(new MapStream(rs, new NullMapper(valid_cols.ToArray())));
rs = new MapStream(rs, new NullMapper(valid_cols.ToArray()));
}
mySetRealSource(rs);
}
@@ -72,9 +71,9 @@ namespace BCG {
public IntSumStream(ViewStream stream, string spaceSeparatedOutputCols, string spaceSeparatedGroupByCols) {
string[] outputCols = spaceSeparatedOutputCols.Split(' ');
string[] groupByCols = spaceSeparatedGroupByCols.Split(' ');
mySetRealSource(AttachStream(new GroupByStream(stream,
mySetRealSource(new GroupByStream(stream,
outputCols.Length == 1 ? new SumGroupByEvaluator<int>(outputCols[0], groupByCols) as IGroupByEvaluator :
new MultiColumnSumGroupByEvaluator<int>(spaceSeparatedOutputCols, spaceSeparatedGroupByCols))));
new MultiColumnSumGroupByEvaluator<int>(spaceSeparatedOutputCols, spaceSeparatedGroupByCols)));
}
}
@@ -83,8 +82,8 @@ namespace BCG {
public class SumStream<T> : ViewStream {
public SumStream(ViewStream stream, string outputCol, string spaceSeparatedInputCols) {
string[] inputCols = spaceSeparatedInputCols.Split(' ');
mySetRealSource(AttachStream(new GroupByStream(stream,
new SumGroupByEvaluator<T>(outputCol, inputCols))));
mySetRealSource(new GroupByStream(stream,
new SumGroupByEvaluator<T>(outputCol, inputCols)));
}
}
}
@@ -22,9 +22,9 @@ namespace BCG
{
if (myUseSuperJoin)
{
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));
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;
mySetRealSource(rs);
}
else
@@ -65,9 +65,9 @@ namespace BCG
{
if (myUseSuperJoin)
{
ViewStream rs = AttachStream(this is LeftJoinStream
? new SuperLeftJoinStream(v1, joinCol1, v2, joinCol2)
: new SuperJoinStream(v1, joinCol1, v2, joinCol2));
ViewStream rs = this is LeftJoinStream ?
new SuperLeftJoinStream(v1, joinCol1, v2, joinCol2) as ViewStream :
new SuperJoinStream(v1, joinCol1, v2, joinCol2) as ViewStream;
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(AttachStream(leftJoin ? new SuperLeftJoinStream(leftStream, leftJoinColArray[0], leftProject,
mySetRealSource(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();
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"));
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");
mySetRealSource(res);
}
}
@@ -15,6 +15,7 @@ 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;
@@ -34,17 +35,22 @@ namespace BCG
: this(NonIncJoinSpec.from_triplet(join_specs), null, 0)
{ }
public NonIncJoinStream(INonIncJoinInlineMapper output_mapper, int throttles_ms,
public NonIncJoinStream(ViewStreamDatabase vsd, INonIncJoinInlineMapper output_mapper, int throttles_ms,
params Triplet<ViewStream, string, string>[] join_specs)
: this(NonIncJoinSpec.from_triplet(join_specs), output_mapper, throttles_ms)
: this(vsd, 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()
@@ -59,7 +65,7 @@ namespace BCG
}
if (_output_mapper != null)
_output_mapper.onMapperChange += _on_mapper_change;
base.myOnFirstSubscriber();
_vsd.addWorker(this);
}
protected override void myOnLastSubscriber()
@@ -68,7 +74,7 @@ namespace BCG
d.Dispose();
if (_output_mapper != null)
_output_mapper.onMapperChange -= _on_mapper_change;
base.myOnLastSubscriber();
_vsd.removeWorker(this);
}
private void _stream_callback()
@@ -52,7 +52,7 @@ namespace BCG {
throw (new Exception("Error occured at line: " + n.ToString() + " " + e.Message));
}
mySetRealSource(AttachStream(new BlueRowCollectionStaticStream(t)));
mySetRealSource(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(AttachStream(new SuperMapStream(vs, mapper)));
mySetRealSource(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(AttachStream(new SuperMapStream(vs, mapper)));
mySetRealSource(new SuperMapStream(vs, mapper));
else {
mapper.onMapperChange += new MapperChangeDelegate(myMapperChanged);
myChangeableMapper = mapper;
@@ -56,6 +56,7 @@ 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();
@@ -65,19 +66,24 @@ namespace bts {
public DataPacketReplayStream(string folder_path, string fire_period)
:
this(folder_path, TimeBasedFilter.parse_duration(fire_period)) {
this(ViewStreamDatabase.instance, folder_path, TimeBasedFilter.parse_duration(fire_period)) {
}
public DataPacketReplayStream(string folder_path, TimeSpan fire_period) {
public DataPacketReplayStream(ViewStreamDatabase vsd, string folder_path, TimeSpan fire_period) {
_vsd = vsd;
_folder_path = folder_path;
_fire_period = fire_period;
}
protected override void myOnFirstSubscriber() {
base.myOnFirstSubscriber();
MasterWorker.getInstance(false).addWorker(this, _fire_period);
_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
@@ -133,4 +139,4 @@ namespace bts {
});
}
}
}
}
@@ -7,7 +7,6 @@ using System.Collections.Generic;
using System;
using System.IO;
using System.Linq;
using Microsoft.Extensions.DependencyInjection;
namespace bts {
public class PubSubTestingStream : ViewStream, IWorker {
@@ -62,18 +61,20 @@ namespace bts {
private readonly int _msg_size;
private readonly TimeSpan _publish_period;
private DateTime _last_publish_time = DateTime.MinValue;
private IPublisherCore? _pub_core_backing;
private PublisherCore _pub_core = new PublisherCore();
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) {
public PubSubTestingStream(int publish_period_msecs, int max_pubs, int payload_size, string tag_name, ViewStreamDatabase vsd) {
_vsd = vsd;
_payload_sz = payload_size;
var schema = BlueDataUtility.form_schema("PubSubTest",
Tuple.Create("id", TypeCode.Int32),
@@ -92,18 +93,18 @@ namespace bts {
_tag_name = tag_name;
}
IPublisherCore? _pub_core =>
_pub_core_backing ??= ActivatorUtilities.CreateInstance<IPublisherCore>(ServiceProvider);
public PubSubTestingStream(int publish_period_seconds, int max_pubs, int msg_size, string tag_name)
: this(publish_period_seconds, max_pubs, msg_size, tag_name, ViewStreamDatabase.instance) { }
protected override void myOnFirstSubscriber() {
base.myOnFirstSubscriber();
_vsd.addWorker(this);
System.Threading.Thread th = new System.Threading.Thread(_subscribe_thread_func);
th.IsBackground = true;
th.Start();
}
protected override void myOnLastSubscriber() {
base.myOnLastSubscriber();
_vsd.removeWorker(this);
_continue_sub_thread = false;
}
@@ -117,7 +118,7 @@ namespace bts {
private void _subscribe_thread_func() {
// subscription is done in another thread
var sub_core = ActivatorUtilities.CreateInstance<SubscriberCore>(ServiceProvider);
SubscriberCore sub_core = new SubscriberCore();
sub_core.subscribe(_tag_name);
while (_continue_sub_thread) {
// TODO recv_id
@@ -151,7 +152,7 @@ namespace bts {
}
var ser_bytes = _last_sent_msg.serialize();
_n_pubs++;
Log.notnull(_pub_core).publish(string.Empty, _tag_name, ser_bytes);
_pub_core.publish(String.Empty, _tag_name, ser_bytes);
_fire_dirty = true;
}
@@ -192,4 +193,4 @@ namespace bts {
}
}
}
}
@@ -0,0 +1,202 @@
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() {
base.myOnFirstSubscriber();
ViewStreamDatabase.instance.addWorker(this);
_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, IParameterizedWorker {
public class SnapshotReplayStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
private ViewStream _snapshot_stream;
private readonly string _sort_col;
private readonly int _buffer_size;
@@ -25,24 +25,22 @@ namespace bts {
public SnapshotReplayStream(ViewStream base_str, string sort_col, int buffer_size, string fire_period)
: this(base_str, sort_col, buffer_size, TimeBasedFilter.parse_duration(fire_period)) { }
: this(ViewStreamDatabase.instance, base_str, sort_col, buffer_size, TimeBasedFilter.parse_duration(fire_period)) { }
public SnapshotReplayStream(ViewStream base_str, string sort_col, int buffer_size, TimeSpan fire_period) {
public SnapshotReplayStream(ViewStreamDatabase vsd, ViewStream base_str, string sort_col, int buffer_size, TimeSpan fire_period) {
_vsd = vsd;
_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;
base.myOnFirstSubscriber();
MasterWorker.getInstance(false).addWorker(this, _fire_period);
}
protected override void myOnLastSubscriber() {
_snapshot_stream.onChange -= _handle_stream_update;
base.myOnLastSubscriber();
}
@@ -76,4 +74,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 = DSTagManager.get_stream_tag(schema.tableName);
var tag = "DS_" + 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(tag);
PubsubServerSingleStreamTagHandler handler = new PubsubServerSingleStreamTagHandler();
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,52 +616,53 @@ namespace BCG {
}
// 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
// };
// }
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
};
}
@@ -686,12 +687,13 @@ 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,11 +3,12 @@ namespace BCG {
public class TickingTestStream : ViewStream {
public TickingTestStream(int ticking_period_msecs)
: this(TimeSpan.FromSeconds(ticking_period_msecs)) { }
:
this(ViewStreamDatabase.instance, TimeSpan.FromSeconds(ticking_period_msecs)) { }
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"));
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");
mySetRealSource(vs);
//vs.onChange += _on_str_update;
}
@@ -27,4 +28,4 @@ namespace BCG {
// myFireEvent(dp_out);
// }
}
}
}
@@ -6,8 +6,6 @@ using System.Security.Cryptography.X509Certificates;
using BCG;
using BCG.Utilities.General;
using Bts.PubSub;
using HtmlAgilityPack;
namespace BCG {
@@ -48,17 +46,19 @@ namespace BCG {
}
}
public class WebScapperStream : ViewStream, IParameterizedWorker {
public class WebScapperStream : ViewStream, IWorker {
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']") { }
public WebScapperStream(string url, string xpath) {
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) {
bts.Log.Gen.info($"WebScapperStream cons url: {url} xpath: {xpath}");
_schema = BlueDataUtility.form_schema($"{url}_{xpath}",
Tuple.Create("id", TypeCode.Int32),
@@ -66,9 +66,16 @@ namespace BCG {
Tuple.Create("line", TypeCode.String));
_url = url;
_xpath = xpath;
_vsd = vsd;
}
TimeSpan IParameterizedWorker.Interval => TimeSpan.FromSeconds(1);
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(1));
}
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
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 = AttachStream(new TimeBasedFilterStream(output_stream, "created_at", ttl));
output_stream = 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(AttachStream(new ObservableViewStream(this)));
mySetRealSource(new ObservableViewStream(this));
}
MongoClient client => clientBacking ??= new(url);
@@ -1,8 +1,6 @@
using BCG;
using BCG.Utilities.General;
using Bts.PubSub;
using MongoDB.Bson;
using MongoDB.Driver;
@@ -12,7 +10,8 @@ using System.Diagnostics;
namespace bts {
[DataStreamUserRespresentation("mongodb_sql")]
public class MongoDBSQLStream : ViewStream, IParameterizedWorker {
public class MongoDBSQLStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
private readonly string _connection_string, _db_name, _collection_name;
private MongoClient _client;
// make this a sorted table
@@ -24,17 +23,28 @@ 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) {
public MongoDBSQLStream(string connection_string, string db_name, string collection_name)
: this(connection_string, db_name, collection_name, ViewStreamDatabase.instance) {
}
public MongoDBSQLStream(string connection_string, string db_name, string collection_name, ViewStreamDatabase vsd) {
bts.Log.Gen.info($"MongoDBStream constructed connection_string: {connection_string} " +
$"db_name: {db_name}");
_connection_string = connection_string;
_db_name = db_name;
_collection_name = collection_name;
_vsd = vsd;
_client = new MongoClient(_connection_string);
_database = _client.GetDatabase(_db_name);
}
TimeSpan IParameterizedWorker.Interval => TimeSpan.FromSeconds(10);
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(10));
}
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
public void doWork() {
_read_mogodb();
@@ -97,7 +107,8 @@ namespace bts {
}
[DataStreamUserRespresentation("mongodb")]
public class MongoDBStream : ViewStream, IParameterizedWorker {
public class MongoDBStream : ViewStream, IWorker {
private ViewStreamDatabase _vsd;
private readonly string _connection_string, _db_name, _collection_name;
private MongoClient _client;
private BlueTable _table;
@@ -107,17 +118,28 @@ 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) {
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) {
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);
}
TimeSpan IParameterizedWorker.Interval => TimeSpan.FromSeconds(10);
protected override void myOnFirstSubscriber() {
MasterWorker.getInstance(false).addWorker(this, TimeSpan.FromSeconds(10));
}
protected override void myOnLastSubscriber() {
MasterWorker.getInstance(false).removeWorker(this);
}
public void doWork() {
_read_mogodb();
@@ -190,12 +212,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(connection_string, db_name, collection_name, timestamp_col, TimeBasedFilter.parse_duration(ttl_duration)) { }
this(ViewStreamDatabase.instance, connection_string, db_name, collection_name, timestamp_col, TimeBasedFilter.parse_duration(ttl_duration)) { }
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));
public MongoDBFlatStream(ViewStreamDatabase vsd, string connection_string, string db_name, string collection_name, string timestamp_col, TimeSpan ttl_duration) {
ViewStream rs = new MongoDBStream(connection_string, db_name, collection_name, vsd);
rs = new KeyValueToRelationalStream(rs, "doc_id", "key", "val");
rs = new DateStandardizationStream(rs, timestamp_col, DateConversion.ISO8601);
//rs = new TimeBasedFilterStream(vsd, rs, timestamp_col, ttl_duration);
mySetRealSource(rs);
}
@@ -207,21 +229,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(connection_string, db_name, collection_name, buffer_sz, TimeBasedFilter.parse_duration(fire_period)) { }
this(ViewStreamDatabase.instance, connection_string, db_name, collection_name, buffer_sz, TimeBasedFilter.parse_duration(fire_period)) { }
public NewsReplayStream(string connection_string, string db_name, string collection_name,
public NewsReplayStream(ViewStreamDatabase vsd, string connection_string, string db_name, string collection_name,
int buffer_sz, TimeSpan fire_period) {
ViewStream r;
r = AttachStream<MongoDBStream>(new(connection_string, db_name, collection_name));
r = AttachStream<KeyValueToRelationalStream>(new(r,
r = new MongoDBStream(connection_string, db_name, collection_name, vsd);
r = new KeyValueToRelationalStream(r,
"doc_id",
"key",
"value"));
r = AttachStream<DateStandardizationStream>(new(r,
"value");
r = new DateStandardizationStream(r,
"created_at",
"ISO8601"));
r = AttachStream<SnapshotReplayStream>(new(r, "created_at", buffer_sz, fire_period));
"ISO8601");
r = new SnapshotReplayStream(vsd, r, "created_at", buffer_sz, fire_period);
mySetRealSource(r);
}
@@ -25,7 +25,7 @@ namespace BCG
myInColumn = inColumn;
myMuxedValues = (object[])muxedValues.Clone();
mySetRealSource(AttachStream(new MapStream(inStream, this)));
mySetRealSource(new MapStream(inStream, this));
}
#region IStreamMapper Members
@@ -8,14 +8,14 @@ namespace BCG
{
public NullToConstantStream(ViewStream v, string field, object constant)
{
var vs = AttachStream<MapStream>(string.IsNullOrEmpty(field)
? new(v, new AllNullToConstantMapper(constant))
: new(v,new NullToConstantMapper(field, constant)));
mySetRealSource(vs);
if(string.IsNullOrEmpty(field))
mySetRealSource(new MapStream(v, new AllNullToConstantMapper(constant)));
else
mySetRealSource (new MapStream(v,new NullToConstantMapper(field, constant)));
}
public NullToConstantStream(ViewStream v, object constant)
{
mySetRealSource(AttachStream(new MapStream(v, new AllNullToConstantMapper(constant))));
mySetRealSource(new MapStream(v, new AllNullToConstantMapper(constant)));
}
}
}
@@ -87,7 +87,7 @@ namespace BCG
public ChangeTableNameStream (ViewStream vs, string name)
{
myNewName = name;
mySetRealSource (AttachStream(new MapStream (vs, this)));
mySetRealSource (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 = AttachStream(new MapStream(vs, this));
_stream = new MapStream(vs, this);
mySetRealSource(_stream);
}
@@ -30,7 +30,7 @@ namespace BCG {
public ProjectStream (ViewStream vs, string colExp, string tableName)
{
if (colExp == "!")
mySetRealSource (AttachStream(new ChangeTableNameStream (vs, tableName)));
mySetRealSource (new ChangeTableNameStream (vs, tableName));
else
{
myVs = vs;
@@ -134,7 +134,7 @@ namespace BCG
{
public CSVReaderStreamMapperStream(ViewStream stream, string output_schema)
{
mySetRealSource(AttachStream(new MapStream(stream, new CSVReaderStreamMapper(output_schema))));
mySetRealSource(new MapStream(stream, new CSVReaderStreamMapper(output_schema)));
}
}
@@ -24,7 +24,6 @@ namespace bts {
var dp_out = BlueDataUtility.build_dp_inc(_rows, myLastValue as DataPacket);
myFireEvent(dp_out);
}
@@ -11,7 +11,13 @@ 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;
@@ -30,12 +36,12 @@ namespace BCG
protected override void myOnFirstSubscriber()
{
myStream.onChange +=new ChangeDelegate(myStreamChange);
base.myOnFirstSubscriber();
myVsd.addWorker(this);
}
protected override void myOnLastSubscriber()
{
base.myOnLastSubscriber();
myVsd.removeWorker(this);
myStream.onChange -= new ChangeDelegate(myStreamChange);
}
#endregion
@@ -16,17 +16,20 @@ namespace BCG
private ViewStream myParentStream;
private ThreadSafeQueue<DataPacket> myQueuedUpdates;
private ViewStreamDatabase myVSD;
public ThreadSafeStream(ViewStream parentStream)
public ThreadSafeStream(ViewStream parent) : this(ViewStreamDatabase.instance, parent) { }
public ThreadSafeStream(ViewStreamDatabase vsd, ViewStream parentStream)
{
myParentStream = parentStream;
myVSD = vsd;
myQueuedUpdates = new ThreadSafeQueue<DataPacket>();
}
protected override void myOnFirstSubscriber()
{
myParentStream.onChange += new ChangeDelegate(myParentStreamChangeHandler);
base.myOnFirstSubscriber();
myVSD.addWorker(this);
}
@@ -34,7 +37,7 @@ namespace BCG
protected override void myOnLastSubscriber()
{
myParentStream.onChange -= new ChangeDelegate(myParentStreamChangeHandler);
base.myOnLastSubscriber();
myVSD.removeWorker(this);
}
void myParentStreamChangeHandler(object newValue)
@@ -50,7 +50,7 @@ namespace BCG {
public DateStandardizationStream(ViewStream vs, string colName, DateConversion conversion) {
myConversionColName = colName;
myConversion = conversion;
mySetRealSource(AttachStream(new MapStream(vs, this)));
mySetRealSource(new MapStream(vs, this));
}
#region IStreamMapper Members
@@ -135,4 +135,4 @@ namespace BCG {
}
}
}
}
}
@@ -15,23 +15,32 @@ namespace BCG
this (offsetInDays,new TimeSpan(0))
{}
public DayStream (int offsetInDays, int h, int m, int s)
public DayStream (int offsetInDays, TimeSpan zeroH)
:
this (offsetInDays, new TimeSpan(0,h,m,s,0))
this (offsetInDays, zeroH, ViewStreamDatabase.instance)
{}
public DayStream (int offsetInDays, TimeSpan zeroHour)
public DayStream (int offsetInDays, int h, int m, int s)
:
this (offsetInDays, new TimeSpan(0,h,m,s,0), ViewStreamDatabase.instance)
{}
public DayStream (int offsetInDays, TimeSpan zeroHour, ViewStreamDatabase db)
{
myOffset = offsetInDays;
myZeroHour = zeroHour;
myDB = db;
}
protected override void myOnFirstSubscriber()
{
base.myOnFirstSubscriber();
myDB.addWorker(this);
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(AttachStream(new MapStream(input, this)));
mySetRealSource(new MapStream(input, this));
}
#region IStreamMapper Members
@@ -1,9 +1,57 @@
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) { }
: base(DateTime.Now, new TimeSpan(0, 0, 0, 0, intervalInms), size, ViewStreamDatabase.instance) { }
protected override DateTime myRoundTime(DateTime dt) { return dt; }
}
@@ -13,9 +61,7 @@ namespace BCG {
new ColumnInfo("id", 0, typeof(int)),
new ColumnInfo("time", 1, typeof(string)));
public DisplayableMillisecondStream(int intervalInMs, int sz) {
ViewStream vs = AttachStream(new MSecondStream(intervalInMs, sz));
vs = AttachStream(new MapStream(vs, this));
mySetRealSource(vs);
mySetRealSource(new MapStream(new MSecondStream(intervalInMs, sz), this));
}
#region IStreamMapper Members
@@ -4,14 +4,22 @@ 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) :
base (zeroTime, new TimeSpan (0,0,0,intervalInSeconds,0), sizeOfStream)
this (zeroTime, intervalInSeconds, sizeOfStream, ViewStreamDatabase.instance)
{}
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,21 +10,36 @@ 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(0)
: this(ViewStreamDatabase.instance, 0)
{ }
public TickStream(int periodMs)
: this(ViewStreamDatabase.instance, periodMs)
{ }
public TickStream(int periodMs)
public TickStream(ViewStreamDatabase vsd, 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,8 +2,6 @@ using System;
using BCG.Utilities.General;
using Bts.DataStreamCore;
namespace BCG {
public abstract class TimeUnitStream : ViewStream, IWorker {
protected TimeSpan myOffset;
@@ -11,13 +9,15 @@ 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) {
public TimeUnitStream(DateTime zeroTime, TimeSpan offSet, int sizeOfStream, ViewStreamDatabase db) {
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,15 +27,17 @@ 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(AttachStream(new MutableAppendStream(v1, v2)));
mySetRealSource(new MutableAppendStream(v1, v2));
else
{
myLeftVs = v1;
@@ -27,7 +27,7 @@ namespace BCG
{
if (myUseNewAppend)
mySetRealSource(AttachStream(new MutableAppendStream(vss)));
mySetRealSource(new MutableAppendStream(vss));
else
{
if (vss.Length == 1)
@@ -36,7 +36,7 @@ namespace BCG
{
AppendStream rv = null;
foreach (ViewStream s in vss)
rv = AttachStream<AppendStream>(rv is null ? new(s) : new(rv, s));
rv = rv == null ? new AppendStream(new ViewStream[] { s }) : new AppendStream(rv, s);
mySetRealSource(rv);
}
else
@@ -112,9 +112,7 @@ namespace BCG
{
public MergeStream (ViewStream v1, ViewStream v2, string[] mergeCols)
{
ViewStream vs = AttachStream(new AppendStream (v1, v2));
vs = AttachStream(new DistinctRowStream (vs, mergeCols));
mySetRealSource(vs);
mySetRealSource (new DistinctRowStream (new AppendStream (v1, v2), mergeCols));
}
}
}
@@ -21,7 +21,7 @@ namespace BCG
{
AppendStreamModified rv = null;
foreach (ViewStream s in vss)
rv = AttachStream<AppendStreamModified>(rv is null ? new (new[] {s}) : new (rv, s));
rv = rv == null ? new AppendStreamModified (new ViewStream[] {s}) : new AppendStreamModified (rv, s);
mySetRealSource (rv);
}
}
@@ -1,3 +1,5 @@
using Bts.DataStreamCore;
using System;
namespace BCG {
@@ -7,17 +9,15 @@ namespace BCG {
/// </summary>
public class CircularPublisher {
private CircularRowCollection myBuffer;
private PubTableBroker myBroker;
private readonly IPubTableBroker myBroker;
private int myBufferSize;
readonly IPublisherCore publisherCore;
public CircularPublisher(int bufferSize, IPublisherCore publisherCore) {
public CircularPublisher(int bufferSize, IPubTableBroker broker) {
myBufferSize = bufferSize;
this.publisherCore = publisherCore;
myBroker = broker;
}
public CircularPublisher(int bufferSize, IPublisherCore publisherCore, IBlueRowCollection rows)
: this(bufferSize, publisherCore) {
public CircularPublisher(int bufferSize, IPubTableBroker broker, IBlueRowCollection rows)
: this(bufferSize, broker) {
publishRows(rows);
}
@@ -26,7 +26,6 @@ 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);
@@ -41,7 +40,6 @@ 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,31 +1,18 @@
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;
IPubTableBroker? _broker_backing;
private readonly PubTableBroker _broker = new PubTableBroker();
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);
}
@@ -41,4 +28,4 @@ namespace BCG {
}
}
}
@@ -8,6 +8,7 @@ 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,94 +1,11 @@
namespace Bts.DataStream.Services;
using BCG;
using bts.api.core;
using Bts.PubSub;
using BCG.Utilities.General;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
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)));
// }
public static class DataStreamServices {
public static IServiceCollection AddRemoteQueryManager(this IServiceCollection services) =>
services.AddSingleton<IWorker, RemoteQueryManager>();
}
@@ -16,12 +16,6 @@ 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) {
@@ -1,7 +0,0 @@
namespace Bts.DataStreamCore;
using BCG;
public interface IConfiguredViewStream {
// ViewStream
}
@@ -17,6 +17,8 @@ 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;
@@ -121,8 +123,8 @@ namespace BCG
protected void mySetRealSource (DataStream src)
{myRealSource = src;}
protected virtual void myOnFirstSubscriber() { }
protected virtual void myOnLastSubscriber () { }
protected virtual void myOnFirstSubscriber() {myFirstSubImplemented = false;}
protected virtual void myOnLastSubscriber () {myLastSubImplemented = false;}
protected virtual void myFireEvent(object o)
{
last_fired_time = DateTime.Now;
@@ -14,7 +14,6 @@
<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>
@@ -1,8 +0,0 @@
namespace Bts.DataStreamCore;
using BCG;
public static class DataStreamCoreExtensions {
public static bool HasSubscribers(this DataStream dataStream) =>
dataStream.numberOfSubscribers > 0;
}
@@ -1,9 +1,6 @@
#nullable enable
namespace BCG {
using bts;
using Bts.DataStream;
using Bts.Extensions.DependencyInjection;
using bts;
using mbts.config;
@@ -11,42 +8,20 @@ namespace BCG {
using System.Collections;
using System.Reflection;
using System.IO;
using System.Collections.Generic;
using Bts.DataStreamCore;
using Bts.PubSub;
public class PrimitiveManager {
Dictionary<string, Type> myTypeLookupTable = new();
private IDictionary myTypeLookupTable = new Hashtable();
private static PrimitiveManager mySelf;
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;
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;
private PrimitiveManager() {
myLoadAllTypes();
}
private void myLoadAllTypes() {
AppDomain currentDomain = AppDomain.CurrentDomain;
currentDomain.AssemblyResolve += myAssemblyResolver;
loadAssembly(GetType()!.Assembly!.FullName!);
currentDomain.AssemblyResolve += new ResolveEventHandler(myAssemblyResolver);
loadAssembly(this.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}");
@@ -60,12 +35,10 @@ namespace BCG {
}
}
static Assembly? myAssemblyResolver(object? sender, ResolveEventArgs args) {
foreach (var a in AppDomain.CurrentDomain.GetAssemblies()) {
if (args.Name.Equals(a.FullName)) {
private static Assembly myAssemblyResolver(object sender, ResolveEventArgs args) {
foreach (Assembly a in AppDomain.CurrentDomain.GetAssemblies())
if (args.Name.Equals(a.FullName))
return a;
}
}
return null;
}
@@ -78,34 +51,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);
var dsatt = type.GetCustomAttribute<DataStreamUserRespresentation>();
if (dsatt is not null) {
isVisible = dsatt.isVisible;
userRep = isVisible ?
dsatt.friendlyName is null ? userRep : dsatt.friendlyName
: null;
}
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;
}
if (isVisible) {
Log.notnull(userRep);
Log.assert_d(!myTypeLookupTable.ContainsKey(userRep), type.FullName!);
Log.assert_d(myTypeLookupTable[userRep] == null, type.FullName);
myTypeLookupTable[userRep] = type;
}
}
}
}
static string createDefaultName(string typeName) {
public static string createDefaultName(string typeName) {
if (!typeName.EndsWith("Stream"))
return typeName;
string retVal = typeName.Remove(typeName.Length - "Stream".Length, "Stream".Length);
@@ -115,14 +88,22 @@ 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 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");
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;
}
/// <summary>
/// Constructs and returns a ViewStream of type corresponding to the friendly name
@@ -131,31 +112,18 @@ 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);
}
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);
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;
}
/// <summary>
@@ -1,7 +0,0 @@
namespace Bts.DataStreamCore;
using BCG.Utilities.General;
using Bts.PubSub;
public interface IViewStreamDatabase : IWorker, IWorkScheduler { }
@@ -1,13 +0,0 @@
#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,6 +21,9 @@ namespace BCG {
protected bool myWaitingForFirstOp = false;
public PubTableBroker()
: this(new PublisherCore()) { }
public PubTableBroker(IPublisherCore publisher) {
_publisher_core = publisher;
Log.Gen.info("PubTableBroker constructed");
@@ -50,13 +53,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,18 +5,21 @@ 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;
}
@@ -25,10 +28,13 @@ 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);
@@ -38,14 +44,17 @@ 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;
@@ -55,16 +64,19 @@ 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" : BlueDataUtility.id_string(_table);
var state_txt = _table == null ? "NULL" : _table.Count.ToString();// 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;
@@ -77,11 +89,13 @@ 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_id_based(_table, table_0, out d_rows, out a_rows);
BlueDataUtility.compute_differences_fast(_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,7 +19,6 @@ 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));
@@ -48,4 +47,4 @@ namespace bts {
private bool _received_create => _schema != null;
}
}
}
@@ -1,8 +1,6 @@
namespace bts {
using BCG;
using Bts.PubSub;
using System;
using System.Collections.Generic;
using System.IO;
@@ -12,13 +10,11 @@ namespace bts {
//byte[] _state = null;
private MemoryStream _str = new MemoryStream();
private BinaryWriter _bw;
public PubsubServerGenericDataHandler() {
_bw = new BinaryWriter(_str);
}
public string Tag => ".*";
public string TagType => ".*";
public string TagType => "*";
public void update(string tag, byte[] msg) {
_str.Position = 0;
@@ -32,8 +28,6 @@ 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
@@ -46,31 +40,24 @@ 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} str_msg: {str_msg} _state: {state_txt}");
Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (pre.1) tag: {tag} sz: {msg.Length} _state: {state_txt}");
switch (str_msg.op) {
case StreamIOMessage.StreamOp.CREATE:
Log.isnull(_state);
//Log.assert(_state == null);
_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} str_msg: {str_msg} state: {_state.Count}");
Log.Gen.info($"PubsubServerSingleStreamTagHandler.update (post) tag: {tag} sz: {msg.Length} state: {_state.Count}");
}
public byte[] get_content(string tag) {
@@ -88,18 +75,17 @@ 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);
@@ -111,21 +97,12 @@ 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(tag);
rv = new PubsubServerSingleStreamTagHandler();
_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(AttachStream(new TableStream("prepub_" + myTag)));
mySetRealSource(new TableStream("prepub_" + myTag, db));
}
catch (MyException e)
{
// BlueEventLog2.instance().logEvent(e.Message, BlueEventLog2Level.Fatal);
mySetRealSource(AttachStream(new TableStream("error_tag")));
mySetRealSource(new TableStream("error_tag"));
}
}
@@ -9,16 +9,13 @@ 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 AddPrimitiveManager(this IServiceCollection services) =>
services.AddSingleton<PrimitiveManager>();
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;
});
}
@@ -1,34 +0,0 @@
// 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;
// }
// }
@@ -1,10 +0,0 @@
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,35 +1,39 @@
#nullable enable
namespace BCG;
using bts;
using Bts.PubSub;
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;
// 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) =>
_table_name = table_name;
public TableStream(string table_name)
:
this(table_name,
ViewStreamDatabase.instance) { }
public override string ToString() =>
$"{GetType()} table: {_table_name}";
protected override void myOnFirstSubscriber() =>
Log.notnull(_ta).on_update += _handle_updates;
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 myOnLastSubscriber() =>
Log.notnull(_ta).on_update -= _handle_updates;
protected override void myOnFirstSubscriber() {
_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 override void myOnLastSubscriber() {
_ta.on_update -= new TableUpdateDelegate(_handle_updates);
}
protected override void OnViewStreamDatabase(IWorkScheduler viewStreamDatabase) {
_ta = (viewStreamDatabase as ViewStreamDatabase)?.getITableAccumulator(_table_name);
Log.notnull(_ta, "ITableAccumulator is null");
base.OnViewStreamDatabase(viewStreamDatabase);
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}";
}
}
}
@@ -1,21 +1,11 @@
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>
@@ -24,10 +14,6 @@ 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;
@@ -39,11 +25,6 @@ namespace BCG
public ViewStream()
{}
#nullable enable
event Action? SubscriptionStarted;
event Action? SubscriptionStopped;
#nullable disable
public IBlueRowCollection rows
{
get
@@ -75,23 +56,6 @@ 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)
@@ -131,53 +95,6 @@ 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,14 +9,16 @@ using System.Threading;
using mbts;
using System.Linq;
using System.Diagnostics;
using Bts.DataStreamCore;
using Bts.PubSub;
using Microsoft.Extensions.Options;
using bts;
namespace BCG {
public class ViewStreamDatabase : IViewStreamDatabase, IDisposable {
public class ViewStreamDatabase : IWorker, 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);
@@ -24,7 +26,6 @@ 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;
@@ -37,16 +38,50 @@ namespace BCG {
private static readonly object myInstanceLockObject = new object();
private readonly object myStateLock = new object();
public ViewStreamDatabase(ISubscriberCore subscriberCore, IOptions<ViewStreamDatabaseConfig> config)
: this(subscriberCore, config.Value) { }
public ViewStreamDatabase()
: this(new SubscriberCore()) {
}
public ViewStreamDatabase(ISubscriberCore subscriberCore, ViewStreamDatabaseConfig config) {
public ViewStreamDatabase(ISubscriberCore subscriberCore) {
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; } }
@@ -73,6 +108,9 @@ 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);
@@ -87,7 +125,11 @@ 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}");
@@ -118,6 +160,8 @@ namespace BCG {
_pending_worker_ops.Clear();
_is_in_do_work = false;
_is_worker_call_in_progress = false;
if (onDoWorkEnd != null)
onDoWorkEnd();
}
}
}
@@ -178,12 +222,6 @@ 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() {
@@ -193,6 +231,11 @@ 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, PrimitiveManager primitiveManager) {
public override string print(int startCol, IList<string> activeBindings) {
string prelude = id + " = ";
string result = prelude + expression.print(startCol + prelude.Length, activeBindings, primitiveManager);
string result = prelude + expression.print(startCol + prelude.Length, activeBindings);
if (activeBindings.IndexOf(id) == -1)
activeBindings.Add(id);
return result;
}
public override ViewStream? myGetView(IDictionary<string, StatementExpression> env, PrimitiveManager primitiveManager) {
var result = expression.getView(env, primitiveManager);
public override ViewStream? myGetView(IDictionary<string, StatementExpression> env) {
var result = expression.getView(env);
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, PrimitiveManager primitiveManager) {
public string print(int startPos, IList<string> activeBindings) {
var sb = new StringBuilder();
foreach (var st in Statements) {
sb.Append(new string(' ', startPos));
sb.Append(st.print(startPos, activeBindings, primitiveManager));
sb.Append(st.print(startPos, activeBindings));
sb.AppendLine(";");
}
sb.AppendLine();
@@ -31,10 +31,10 @@ public class DataStreamExpression {
return sb.ToString();
}
public ViewStream? getView(IDictionary<string, StatementExpression> env, PrimitiveManager primitiveManager) {
public ViewStream? getView(IDictionary<string, StatementExpression> env) {
for (int i = 0; i < Statements.Count - 1; i++) {
Statements[i].getView(env, primitiveManager);
Statements[i].getView(env);
}
return Statements[^1].getView(env, primitiveManager);
return Statements[^1].getView(env);
}
}
Loaded 100 of 208 files, more files were not shown because too many files have changed in this diff. Show more