241 lines
11 KiB
C#
241 lines
11 KiB
C#
using Google.Protobuf.WellKnownTypes;
|
|
using Google.Protobuf;
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text;
|
|
using System.Threading.Tasks;
|
|
using System.Reactive.Linq;
|
|
using System.IO;
|
|
using System.Reactive.Disposables;
|
|
using mbts;
|
|
using System.Threading;
|
|
using System.Runtime.InteropServices;
|
|
|
|
|
|
namespace bts.api.core {
|
|
|
|
internal class QuickbusClientChannel : ClientChannel {
|
|
IntPtr _client;
|
|
|
|
public QuickbusClientChannel(ServiceContext context, string type_name, string qualifier) : base(context) {
|
|
Log.QBus.diag($"Created QuickbusClientChannel");
|
|
_client = ApiNativeMethods.api_client_create(context.native_context, type_name, qualifier);
|
|
System.Diagnostics.Trace.Assert(_client != IntPtr.Zero);
|
|
}
|
|
|
|
public override void Dispose() {
|
|
Log.QBus.diag($"Disposed QuickbusClientChannel");
|
|
ApiNativeMethods.api_client_destroy(_client);
|
|
_client = IntPtr.Zero;
|
|
}
|
|
|
|
|
|
int _req_id = 0;
|
|
Dictionary<int, PBCallbackWrapper> _keepalive_cache = new();
|
|
|
|
unsafe public override Task<Result<Response>> send<Request, Response>(int method_id, Request request) {
|
|
var id = Interlocked.Increment(ref _req_id);
|
|
var header = new RequestHeader() {
|
|
Method = method_id,
|
|
RequestId = id,
|
|
Token = _context.token.HasValue ? new AuthToken() { Token = _context.token?.Token } : null
|
|
};
|
|
var buffer = header.serialize(request);
|
|
var tcs = new TaskCompletionSource<Result<Response>>();
|
|
ApiNativeMethods.raw_pb_callback callback = (rbuffer, rlength, _) => {
|
|
var res = ApiUtils.deserialize<ResponseHeader, Response>(rbuffer, (long)rlength);
|
|
if (res.header.Code == ApiError.Success) {
|
|
tcs.TrySetResult(new Result<Response>(res.body));
|
|
}
|
|
else {
|
|
tcs.TrySetResult(new Result<Response>(res.header.Code));
|
|
}
|
|
lock (_keepalive_cache) {
|
|
_keepalive_cache.Remove(id);
|
|
}
|
|
};
|
|
System.Diagnostics.Trace.Assert(_client != IntPtr.Zero, "Channel may have been disposed");
|
|
var cb_wrapper = new PBCallbackWrapper(callback);
|
|
lock (_keepalive_cache) {
|
|
_keepalive_cache.Add(id, cb_wrapper);
|
|
}
|
|
ApiNativeMethods.api_client_send(
|
|
_client, buffer, buffer.Length, cb_wrapper.Callback, IntPtr.Zero);
|
|
// Return a task that will free the handle once the task is completed
|
|
return tcs.Task.ContinueWith(task => {
|
|
cb_wrapper.Free(); // Ensure the GCHandle is freed when the callback is done
|
|
return task.Result;
|
|
});
|
|
}
|
|
|
|
public override IObservable<RawUpdate<TOut>> subscribe<TParse,TOut>(int method_id, string method, string qualifier, MessageParser<TParse> parser, Func<TParse,TOut> converter) {
|
|
var result = Observable.Create<RawUpdate<TOut>>(async observer => {
|
|
IntPtr temp_subscription = IntPtr.Zero;
|
|
void unsubscribe_temp_sub() {
|
|
// Unsubscribe late to keep qb refcount > 0 continuously
|
|
if (temp_subscription != IntPtr.Zero) ApiNativeMethods.api_client_unsubscribe(_client, temp_subscription);
|
|
}
|
|
|
|
// 'tcs' and 'temp_callback' are defined in the same scope as 'temp_subscription'
|
|
// (rather than in the 'while loop' scope) to prevent premature garbage collection
|
|
TaskCompletionSource<bool> tcs = null;
|
|
ApiNativeMethods.raw_subscription_callback temp_callback =
|
|
(event_type, changes, change_count, unused) => {
|
|
tcs?.TrySetResult(true);
|
|
};
|
|
|
|
while (true) {
|
|
var response = await check_stream_access(method_id, qualifier);
|
|
if (response.success) {
|
|
if (response.value.Authorized) {
|
|
// success. Break loop and subscribe
|
|
break;
|
|
}
|
|
else {
|
|
// Not authorized. return empty stream
|
|
unsubscribe_temp_sub();
|
|
observer.OnCompleted();
|
|
return System.Reactive.Disposables.Disposable.Empty;
|
|
}
|
|
}
|
|
else {
|
|
// Wait for stream to fire once before trying again
|
|
unsubscribe_temp_sub();
|
|
tcs = new TaskCompletionSource<bool>();
|
|
temp_subscription = ApiNativeMethods.api_client_subscribe(
|
|
_client, method, qualifier, temp_callback, IntPtr.Zero);
|
|
await tcs.Task;
|
|
}
|
|
}
|
|
|
|
// Authorized. Stream events
|
|
unsubscribe_temp_sub();
|
|
// Ensure 'temp_callback' and 'tcs' aren't Garbage Collected before 'temp_subscription'
|
|
// is disposed
|
|
GC.KeepAlive(temp_callback);
|
|
GC.KeepAlive(tcs);
|
|
return SubscribeNative(method, qualifier, parser, converter, observer);
|
|
});
|
|
Log.QBus.diag($"Subscribing for sub: '{method}/{qualifier}'");
|
|
return result;
|
|
}
|
|
private Task<Result<StreamAccessResponse>> check_stream_access(int method_id, string qualifier) {
|
|
var request = new StreamAccessRequest() { Qualifier = qualifier ?? "" };
|
|
return send<StreamAccessRequest, StreamAccessResponse>(method_id, request);
|
|
}
|
|
|
|
private unsafe IDisposable SubscribeNative<TParse, TOut>(string method, string qualifier, MessageParser<TParse> parser, Func<TParse, TOut> converter, IObserver<RawUpdate<TOut>> observer) where TParse : IMessage<TParse> {
|
|
bool first_update = true;
|
|
// Memory optimization. There is only one update per subscription.
|
|
RawUpdate<TOut> the_update = new RawUpdate<TOut>();
|
|
ApiNativeMethods.raw_subscription_callback cb = (event_type, changes, change_count, unused) => {
|
|
the_update.Clear();
|
|
try {
|
|
if (event_type == ApiNativeMethods.EventType.TABLE_UPDATED) {
|
|
if (changes != null) {
|
|
foreach (var change in changes) {
|
|
switch (change.type) {
|
|
case ApiNativeMethods.ChangeType.REMOVE:
|
|
the_update.removed_keys.Add(change.key);
|
|
break;
|
|
case ApiNativeMethods.ChangeType.REPLACE:
|
|
the_update.changed_items.Add(converter(parser.ParseFrom(new UnmanagedMemoryStream(change.buffer, change.buffer_len))));
|
|
the_update.changed_keys.Add(change.key);
|
|
break;
|
|
case ApiNativeMethods.ChangeType.CREATE:
|
|
the_update.added_items.Add(converter(parser.ParseFrom(new UnmanagedMemoryStream(change.buffer, change.buffer_len))));
|
|
the_update.added_keys.Add(change.key);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
observer.OnNext(the_update);
|
|
}
|
|
else if (event_type == ApiNativeMethods.EventType.TABLE_CREATED) {
|
|
if (!first_update) {
|
|
observer.OnNext(the_update.MarkAsReset());
|
|
}
|
|
first_update = false;
|
|
}
|
|
}
|
|
catch (Exception e) {
|
|
ApiNativeMethods.api_client_log_error($"Exception in subscription callback for sub: '{method}/{qualifier}' \n--\n '{e}'");
|
|
// Sleep to let the log message get asyncronously written to disk
|
|
Thread.Sleep(100);
|
|
Environment.Exit(-1);
|
|
}
|
|
};
|
|
var subscription = ApiNativeMethods.api_client_subscribe(_client, method, qualifier, cb, IntPtr.Zero);
|
|
return FinalizeableAnonymousDisposable.Create((disposing) => {
|
|
Log.QBus.diag($"Unsubscribing for sub: '{method}/{qualifier}'");
|
|
ApiNativeMethods.api_client_unsubscribe(_client, subscription);
|
|
subscription = IntPtr.Zero;
|
|
GC.KeepAlive(cb);
|
|
cb = null;
|
|
});
|
|
}
|
|
|
|
private class FinalizeableAnonymousDisposable : IDisposable {
|
|
private Action<bool> _action;
|
|
public FinalizeableAnonymousDisposable(Action<bool> dispose) {
|
|
_action = dispose;
|
|
}
|
|
public static FinalizeableAnonymousDisposable Create(Action<bool> dispose) { return new FinalizeableAnonymousDisposable(dispose); }
|
|
~FinalizeableAnonymousDisposable() {
|
|
Dispose(false);
|
|
}
|
|
public void Dispose() {
|
|
GC.SuppressFinalize(this);
|
|
Dispose(true);
|
|
}
|
|
|
|
protected virtual void Dispose(bool disposing) {
|
|
var temp = _action;
|
|
_action = null;
|
|
temp?.Invoke(disposing);
|
|
}
|
|
}
|
|
|
|
public override bool is_sync_call_ok() {
|
|
return !ApiNativeMethods.api_client_is_on_qb_thread(_client);
|
|
}
|
|
|
|
/// <summary>
|
|
/// A simple wrapper that can be used to prevent the premature garbage collection of a
|
|
/// <see cref="ApiNativeMethods.raw_pb_callback"/> delegate.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// This is part of a workaround for an apparent bug in csharp's GarbageCollector, where the
|
|
/// final reference to a delegate instance mustn't be removed until any and all execution has
|
|
/// finished (even if the reference is removed by the delegate's last statemwent, a
|
|
/// garbage-collector race may yield the infamous error, "A callback was made on a garbage
|
|
/// collected delegate of type...").
|
|
/// </remarks>
|
|
class PBCallbackWrapper
|
|
{
|
|
public readonly ApiNativeMethods.raw_pb_callback Callback;
|
|
private GCHandle _handle;
|
|
|
|
public PBCallbackWrapper(ApiNativeMethods.raw_pb_callback callback)
|
|
{
|
|
// Pin the callback to prevent garbage collection
|
|
Log.QBus.diag($"Created PBCallbackWrapper");
|
|
_handle = GCHandle.Alloc(callback);
|
|
Callback = callback;
|
|
}
|
|
|
|
public void Free()
|
|
{
|
|
// Free the GCHandle when the delegate is no longer needed
|
|
Log.QBus.diag($"Deleting PBCallbackWrapper");
|
|
|
|
if (_handle.IsAllocated)
|
|
{
|
|
_handle.Free();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|