feat(client): streamEvents() wraps server-streaming audit log
Some checks failed
Security / Security check (push) Failing after 2s
Some checks failed
Security / Security check (push) Failing after 2s
Adds `HubClient.streamEvents(backfill:, types:)` returning a `Stream<LoggedEvent>` that fronts the new `HubAdmin/StreamEvents` RPC. Each event arrives the instant the hub appends it; cancel the subscription to close the RPC. `backfill` defaults to 50 historical events (oldest-last for chronological reception). `RESOURCE_EXHAUSTED` from the server means the receiver fell behind — reconnect with a fresh backfill to resync. Bindings regenerated from fai_platform/proto. dart analyze + dart test green (4 tests). Signed-off-by: flemming-it <stefan.a.flemming@googlemail.com>
This commit is contained in:
parent
549f9df5c6
commit
c90a88481c
4 changed files with 116 additions and 0 deletions
|
|
@ -864,6 +864,64 @@ class EventLogResponse extends $pb.GeneratedMessage {
|
||||||
$pb.PbList<LoggedEvent> get events => $_getList(0);
|
$pb.PbList<LoggedEvent> get events => $_getList(0);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
class StreamEventsRequest extends $pb.GeneratedMessage {
|
||||||
|
factory StreamEventsRequest({
|
||||||
|
$core.int? backfill,
|
||||||
|
$core.Iterable<$core.String>? eventTypes,
|
||||||
|
}) {
|
||||||
|
final $result = create();
|
||||||
|
if (backfill != null) {
|
||||||
|
$result.backfill = backfill;
|
||||||
|
}
|
||||||
|
if (eventTypes != null) {
|
||||||
|
$result.eventTypes.addAll(eventTypes);
|
||||||
|
}
|
||||||
|
return $result;
|
||||||
|
}
|
||||||
|
StreamEventsRequest._() : super();
|
||||||
|
factory StreamEventsRequest.fromBuffer($core.List<$core.int> i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromBuffer(i, r);
|
||||||
|
factory StreamEventsRequest.fromJson($core.String i, [$pb.ExtensionRegistry r = $pb.ExtensionRegistry.EMPTY]) => create()..mergeFromJson(i, r);
|
||||||
|
|
||||||
|
static final $pb.BuilderInfo _i = $pb.BuilderInfo(_omitMessageNames ? '' : 'StreamEventsRequest', package: const $pb.PackageName(_omitMessageNames ? '' : 'fai.v1'), createEmptyInstance: create)
|
||||||
|
..a<$core.int>(1, _omitFieldNames ? '' : 'backfill', $pb.PbFieldType.O3)
|
||||||
|
..pPS(2, _omitFieldNames ? '' : 'eventTypes')
|
||||||
|
..hasRequiredFields = false
|
||||||
|
;
|
||||||
|
|
||||||
|
@$core.Deprecated('See https://github.com/google/protobuf.dart/issues/998.')
|
||||||
|
StreamEventsRequest clone() => StreamEventsRequest()..mergeFromMessage(this);
|
||||||
|
@$core.Deprecated('See https://github.com/google/protobuf.dart/issues/998.')
|
||||||
|
StreamEventsRequest copyWith(void Function(StreamEventsRequest) updates) => super.copyWith((message) => updates(message as StreamEventsRequest)) as StreamEventsRequest;
|
||||||
|
|
||||||
|
$pb.BuilderInfo get info_ => _i;
|
||||||
|
|
||||||
|
@$core.pragma('dart2js:noInline')
|
||||||
|
static StreamEventsRequest create() => StreamEventsRequest._();
|
||||||
|
StreamEventsRequest createEmptyInstance() => create();
|
||||||
|
static $pb.PbList<StreamEventsRequest> createRepeated() => $pb.PbList<StreamEventsRequest>();
|
||||||
|
@$core.pragma('dart2js:noInline')
|
||||||
|
static StreamEventsRequest getDefault() => _defaultInstance ??= $pb.GeneratedMessage.$_defaultFor<StreamEventsRequest>(create);
|
||||||
|
static StreamEventsRequest? _defaultInstance;
|
||||||
|
|
||||||
|
/// Backfill size before the stream switches to live updates.
|
||||||
|
/// 0 means "no backfill, only new events". Defaults to 50 when
|
||||||
|
/// unset; capped at 1000 server-side to bound the initial buffer.
|
||||||
|
@$pb.TagNumber(1)
|
||||||
|
$core.int get backfill => $_getIZ(0);
|
||||||
|
@$pb.TagNumber(1)
|
||||||
|
set backfill($core.int v) { $_setSignedInt32(0, v); }
|
||||||
|
@$pb.TagNumber(1)
|
||||||
|
$core.bool hasBackfill() => $_has(0);
|
||||||
|
@$pb.TagNumber(1)
|
||||||
|
void clearBackfill() => $_clearField(1);
|
||||||
|
|
||||||
|
/// Optional event_type filter; when empty, all types are streamed.
|
||||||
|
/// The filter applies to both the backfill prefix and the live
|
||||||
|
/// tail.
|
||||||
|
@$pb.TagNumber(2)
|
||||||
|
$pb.PbList<$core.String> get eventTypes => $_getList(1);
|
||||||
|
}
|
||||||
|
|
||||||
class LoggedEvent extends $pb.GeneratedMessage {
|
class LoggedEvent extends $pb.GeneratedMessage {
|
||||||
factory LoggedEvent({
|
factory LoggedEvent({
|
||||||
$core.String? eventId,
|
$core.String? eventId,
|
||||||
|
|
|
||||||
|
|
@ -139,6 +139,10 @@ class HubAdminClient extends $grpc.Client {
|
||||||
'/fai.v1.HubAdmin/EventLog',
|
'/fai.v1.HubAdmin/EventLog',
|
||||||
($0.EventLogRequest value) => value.writeToBuffer(),
|
($0.EventLogRequest value) => value.writeToBuffer(),
|
||||||
($core.List<$core.int> value) => $0.EventLogResponse.fromBuffer(value));
|
($core.List<$core.int> value) => $0.EventLogResponse.fromBuffer(value));
|
||||||
|
static final _$streamEvents = $grpc.ClientMethod<$0.StreamEventsRequest, $0.LoggedEvent>(
|
||||||
|
'/fai.v1.HubAdmin/StreamEvents',
|
||||||
|
($0.StreamEventsRequest value) => value.writeToBuffer(),
|
||||||
|
($core.List<$core.int> value) => $0.LoggedEvent.fromBuffer(value));
|
||||||
static final _$saveFlow = $grpc.ClientMethod<$0.SaveFlowRequest, $0.SaveFlowResponse>(
|
static final _$saveFlow = $grpc.ClientMethod<$0.SaveFlowRequest, $0.SaveFlowResponse>(
|
||||||
'/fai.v1.HubAdmin/SaveFlow',
|
'/fai.v1.HubAdmin/SaveFlow',
|
||||||
($0.SaveFlowRequest value) => value.writeToBuffer(),
|
($0.SaveFlowRequest value) => value.writeToBuffer(),
|
||||||
|
|
@ -315,6 +319,15 @@ class HubAdminClient extends $grpc.Client {
|
||||||
return $createUnaryCall(_$eventLog, request, options: options);
|
return $createUnaryCall(_$eventLog, request, options: options);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Server-streaming variant of EventLog. The hub first emits up
|
||||||
|
/// to `limit` historical events (newest-last so clients receive
|
||||||
|
/// them in chronological order), then keeps the stream open and
|
||||||
|
/// forwards every freshly-appended event in real time. Reads
|
||||||
|
/// require the Read scope when auth is on.
|
||||||
|
$grpc.ResponseStream<$0.LoggedEvent> streamEvents($0.StreamEventsRequest request, {$grpc.CallOptions? options}) {
|
||||||
|
return $createStreamingCall(_$streamEvents, $async.Stream.fromIterable([request]), options: options);
|
||||||
|
}
|
||||||
|
|
||||||
/// Saved-flow management. A saved flow is a YAML file under
|
/// Saved-flow management. A saved flow is a YAML file under
|
||||||
/// <data_dir>/flows/<name>.yaml; it can be re-run by name later.
|
/// <data_dir>/flows/<name>.yaml; it can be re-run by name later.
|
||||||
$grpc.ResponseFuture<$0.SaveFlowResponse> saveFlow($0.SaveFlowRequest request, {$grpc.CallOptions? options}) {
|
$grpc.ResponseFuture<$0.SaveFlowResponse> saveFlow($0.SaveFlowRequest request, {$grpc.CallOptions? options}) {
|
||||||
|
|
@ -619,6 +632,13 @@ abstract class HubAdminServiceBase extends $grpc.Service {
|
||||||
false,
|
false,
|
||||||
($core.List<$core.int> value) => $0.EventLogRequest.fromBuffer(value),
|
($core.List<$core.int> value) => $0.EventLogRequest.fromBuffer(value),
|
||||||
($0.EventLogResponse value) => value.writeToBuffer()));
|
($0.EventLogResponse value) => value.writeToBuffer()));
|
||||||
|
$addMethod($grpc.ServiceMethod<$0.StreamEventsRequest, $0.LoggedEvent>(
|
||||||
|
'StreamEvents',
|
||||||
|
streamEvents_Pre,
|
||||||
|
false,
|
||||||
|
true,
|
||||||
|
($core.List<$core.int> value) => $0.StreamEventsRequest.fromBuffer(value),
|
||||||
|
($0.LoggedEvent value) => value.writeToBuffer()));
|
||||||
$addMethod($grpc.ServiceMethod<$0.SaveFlowRequest, $0.SaveFlowResponse>(
|
$addMethod($grpc.ServiceMethod<$0.SaveFlowRequest, $0.SaveFlowResponse>(
|
||||||
'SaveFlow',
|
'SaveFlow',
|
||||||
saveFlow_Pre,
|
saveFlow_Pre,
|
||||||
|
|
@ -903,6 +923,10 @@ abstract class HubAdminServiceBase extends $grpc.Service {
|
||||||
return eventLog($call, await $request);
|
return eventLog($call, await $request);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
$async.Stream<$0.LoggedEvent> streamEvents_Pre($grpc.ServiceCall $call, $async.Future<$0.StreamEventsRequest> $request) async* {
|
||||||
|
yield* streamEvents($call, await $request);
|
||||||
|
}
|
||||||
|
|
||||||
$async.Future<$0.SaveFlowResponse> saveFlow_Pre($grpc.ServiceCall $call, $async.Future<$0.SaveFlowRequest> $request) async {
|
$async.Future<$0.SaveFlowResponse> saveFlow_Pre($grpc.ServiceCall $call, $async.Future<$0.SaveFlowRequest> $request) async {
|
||||||
return saveFlow($call, await $request);
|
return saveFlow($call, await $request);
|
||||||
}
|
}
|
||||||
|
|
@ -1059,6 +1083,7 @@ abstract class HubAdminServiceBase extends $grpc.Service {
|
||||||
$async.Future<$0.ModuleInfoResponse> moduleInfo($grpc.ServiceCall call, $0.ModuleInfoRequest request);
|
$async.Future<$0.ModuleInfoResponse> moduleInfo($grpc.ServiceCall call, $0.ModuleInfoRequest request);
|
||||||
$async.Future<$0.CheckPermissionResponse> checkPermission($grpc.ServiceCall call, $0.CheckPermissionRequest request);
|
$async.Future<$0.CheckPermissionResponse> checkPermission($grpc.ServiceCall call, $0.CheckPermissionRequest request);
|
||||||
$async.Future<$0.EventLogResponse> eventLog($grpc.ServiceCall call, $0.EventLogRequest request);
|
$async.Future<$0.EventLogResponse> eventLog($grpc.ServiceCall call, $0.EventLogRequest request);
|
||||||
|
$async.Stream<$0.LoggedEvent> streamEvents($grpc.ServiceCall call, $0.StreamEventsRequest request);
|
||||||
$async.Future<$0.SaveFlowResponse> saveFlow($grpc.ServiceCall call, $0.SaveFlowRequest request);
|
$async.Future<$0.SaveFlowResponse> saveFlow($grpc.ServiceCall call, $0.SaveFlowRequest request);
|
||||||
$async.Future<$0.DeleteFlowResponse> deleteFlow($grpc.ServiceCall call, $0.DeleteFlowRequest request);
|
$async.Future<$0.DeleteFlowResponse> deleteFlow($grpc.ServiceCall call, $0.DeleteFlowRequest request);
|
||||||
$async.Future<$0.FlowList> listFlows($grpc.ServiceCall call, $1.Empty request);
|
$async.Future<$0.FlowList> listFlows($grpc.ServiceCall call, $1.Empty request);
|
||||||
|
|
|
||||||
|
|
@ -245,6 +245,20 @@ final $typed_data.Uint8List eventLogResponseDescriptor = $convert.base64Decode(
|
||||||
'ChBFdmVudExvZ1Jlc3BvbnNlEisKBmV2ZW50cxgBIAMoCzITLmZhaS52MS5Mb2dnZWRFdmVudF'
|
'ChBFdmVudExvZ1Jlc3BvbnNlEisKBmV2ZW50cxgBIAMoCzITLmZhaS52MS5Mb2dnZWRFdmVudF'
|
||||||
'IGZXZlbnRz');
|
'IGZXZlbnRz');
|
||||||
|
|
||||||
|
@$core.Deprecated('Use streamEventsRequestDescriptor instead')
|
||||||
|
const StreamEventsRequest$json = {
|
||||||
|
'1': 'StreamEventsRequest',
|
||||||
|
'2': [
|
||||||
|
{'1': 'backfill', '3': 1, '4': 1, '5': 5, '10': 'backfill'},
|
||||||
|
{'1': 'event_types', '3': 2, '4': 3, '5': 9, '10': 'eventTypes'},
|
||||||
|
],
|
||||||
|
};
|
||||||
|
|
||||||
|
/// Descriptor for `StreamEventsRequest`. Decode as a `google.protobuf.DescriptorProto`.
|
||||||
|
final $typed_data.Uint8List streamEventsRequestDescriptor = $convert.base64Decode(
|
||||||
|
'ChNTdHJlYW1FdmVudHNSZXF1ZXN0EhoKCGJhY2tmaWxsGAEgASgFUghiYWNrZmlsbBIfCgtldm'
|
||||||
|
'VudF90eXBlcxgCIAMoCVIKZXZlbnRUeXBlcw==');
|
||||||
|
|
||||||
@$core.Deprecated('Use loggedEventDescriptor instead')
|
@$core.Deprecated('Use loggedEventDescriptor instead')
|
||||||
const LoggedEvent$json = {
|
const LoggedEvent$json = {
|
||||||
'1': 'LoggedEvent',
|
'1': 'LoggedEvent',
|
||||||
|
|
|
||||||
|
|
@ -158,6 +158,25 @@ class HubClient {
|
||||||
return r.events;
|
return r.events;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Server-streaming audit-log subscription. The hub first
|
||||||
|
/// replays up to [backfill] historical events (oldest-last so
|
||||||
|
/// the receiver sees them in chronological order), then keeps
|
||||||
|
/// the stream open and forwards every newly-appended event as
|
||||||
|
/// soon as it lands.
|
||||||
|
///
|
||||||
|
/// Cancel the returned [Stream]'s subscription to close the
|
||||||
|
/// RPC. If the server reports `RESOURCE_EXHAUSTED` ("stream
|
||||||
|
/// lagged"), reconnect with backfill to resync.
|
||||||
|
Stream<LoggedEvent> streamEvents({
|
||||||
|
int backfill = 50,
|
||||||
|
List<String> types = const [],
|
||||||
|
}) {
|
||||||
|
final req = pb.StreamEventsRequest()
|
||||||
|
..backfill = backfill
|
||||||
|
..eventTypes.addAll(types);
|
||||||
|
return _admin.streamEvents(req);
|
||||||
|
}
|
||||||
|
|
||||||
/// Pending approvals filtered by status. Defaults to all.
|
/// Pending approvals filtered by status. Defaults to all.
|
||||||
Future<List<PendingApprovalEntry>> listApprovals({
|
Future<List<PendingApprovalEntry>> listApprovals({
|
||||||
List<String> statuses = const [],
|
List<String> statuses = const [],
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue