diff --git a/cmd/backend/hermes.json b/cmd/backend/hermes.json new file mode 100644 index 0000000..d9a2b6b --- /dev/null +++ b/cmd/backend/hermes.json @@ -0,0 +1,172 @@ +[ + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "127.0.0.1", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + }, + { + "name": "TimescaleDB", + "provider": "TimescaleDB", + "settings": { + "host": "localhost", + "user": "postgres", + "password": "password", + "database": "hermes" + } + } +] \ No newline at end of file diff --git a/lib/python/hermes_pb2_grpc.py b/lib/python/hermes_pb2_grpc.py index c916423..e4706c1 100644 --- a/lib/python/hermes_pb2_grpc.py +++ b/lib/python/hermes_pb2_grpc.py @@ -176,11 +176,21 @@ def __init__(self, channel): request_serializer=bus__pb2.BusFilter.SerializeToString, response_deserializer=bus__pb2.SourcedEvent.FromString, _registered_method=True) + self.EmitEvent = channel.stream_unary( + '/Api/EmitEvent', + request_serializer=bus__pb2.SourcedEvent.SerializeToString, + response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + _registered_method=True) self.SubTelemetry = channel.unary_stream( '/Api/SubTelemetry', request_serializer=bus__pb2.BusFilter.SerializeToString, response_deserializer=bus__pb2.SourcedTelemetry.FromString, _registered_method=True) + self.EmitTelemetry = channel.stream_unary( + '/Api/EmitTelemetry', + request_serializer=bus__pb2.SourcedTelemetry.SerializeToString, + response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + _registered_method=True) self.SubFileDownlink = channel.unary_stream( '/Api/SubFileDownlink', request_serializer=bus__pb2.BusFilter.SerializeToString, @@ -452,6 +462,13 @@ def SubEvent(self, request, context): context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') + def EmitEvent(self, request_iterator, context): + """Emit to the event message bus + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + def SubTelemetry(self, request, context): """Subscribe to the telemetry message bus """ @@ -459,6 +476,13 @@ def SubTelemetry(self, request, context): context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') + def EmitTelemetry(self, request_iterator, context): + """Emit to the telemetry message bus + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + def SubFileDownlink(self, request, context): """Subscribe to the file downlink message bus """ @@ -618,11 +642,21 @@ def add_ApiServicer_to_server(servicer, server): request_deserializer=bus__pb2.BusFilter.FromString, response_serializer=bus__pb2.SourcedEvent.SerializeToString, ), + 'EmitEvent': grpc.stream_unary_rpc_method_handler( + servicer.EmitEvent, + request_deserializer=bus__pb2.SourcedEvent.FromString, + response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + ), 'SubTelemetry': grpc.unary_stream_rpc_method_handler( servicer.SubTelemetry, request_deserializer=bus__pb2.BusFilter.FromString, response_serializer=bus__pb2.SourcedTelemetry.SerializeToString, ), + 'EmitTelemetry': grpc.stream_unary_rpc_method_handler( + servicer.EmitTelemetry, + request_deserializer=bus__pb2.SourcedTelemetry.FromString, + response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + ), 'SubFileDownlink': grpc.unary_stream_rpc_method_handler( servicer.SubFileDownlink, request_deserializer=bus__pb2.BusFilter.FromString, @@ -1379,6 +1413,33 @@ def SubEvent(request, metadata, _registered_method=True) + @staticmethod + def EmitEvent(request_iterator, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.stream_unary( + request_iterator, + target, + '/Api/EmitEvent', + bus__pb2.SourcedEvent.SerializeToString, + google_dot_protobuf_dot_empty__pb2.Empty.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + @staticmethod def SubTelemetry(request, target, @@ -1406,6 +1467,33 @@ def SubTelemetry(request, metadata, _registered_method=True) + @staticmethod + def EmitTelemetry(request_iterator, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.stream_unary( + request_iterator, + target, + '/Api/EmitTelemetry', + bus__pb2.SourcedTelemetry.SerializeToString, + google_dot_protobuf_dot_empty__pb2.Empty.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + @staticmethod def SubFileDownlink(request, target, diff --git a/nasa-hermes-datasource/go.mod b/nasa-hermes-datasource/go.mod index 27977df..3446803 100644 --- a/nasa-hermes-datasource/go.mod +++ b/nasa-hermes-datasource/go.mod @@ -6,6 +6,9 @@ require ( github.com/DATA-DOG/go-sqlmock v1.5.2 github.com/grafana/grafana-plugin-sdk-go v0.292.2 github.com/lib/pq v1.12.3 + github.com/nasa/hermes v0.0.0 + google.golang.org/grpc v1.81.1 + google.golang.org/protobuf v1.36.11 ) require ( @@ -76,6 +79,7 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect go.opentelemetry.io/otel/sdk v1.44.0 // indirect + go.opentelemetry.io/otel/sdk/metric v1.44.0 // indirect go.opentelemetry.io/otel/trace v1.44.0 // indirect go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect @@ -84,7 +88,7 @@ require ( golang.org/x/sys v0.45.0 // indirect golang.org/x/text v0.37.0 // indirect google.golang.org/genproto v0.0.0-20210630183607-d20f26d13c79 // indirect - google.golang.org/grpc v1.81.1 // indirect - google.golang.org/protobuf v1.36.11 // indirect gopkg.in/fsnotify/fsnotify.v1 v1.4.7 // indirect ) + +replace github.com/nasa/hermes => ../ diff --git a/nasa-hermes-datasource/pkg/plugin/timescale_integration_test.go b/nasa-hermes-datasource/pkg/plugin/timescale_integration_test.go new file mode 100644 index 0000000..57fd471 --- /dev/null +++ b/nasa-hermes-datasource/pkg/plugin/timescale_integration_test.go @@ -0,0 +1,296 @@ +package plugin + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "math/rand" + "net" + "os/exec" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/types/known/timestamppb" + + "github.com/grafana/grafana-plugin-sdk-go/backend" + hermesGrpc "github.com/nasa/hermes/pkg/grpc" + pb "github.com/nasa/hermes/pkg/pb" +) + +const ( + timescaleConnStr = "postgres://postgres:password@localhost:5432/hermes?sslmode=disable" + hermesGrpcConnStr = "localhost:50051" +) + +func BenchmarkTimescaleQueries(b *testing.B) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + hermesConn := startHermesBackend(b, ctx) + defer func() { _ = hermesConn.Close() }() + + hermesClient := hermesGrpc.NewApiClient(hermesConn) + + db := startTimescaleGrafana(b, ctx, hermesClient) + defer func() { _ = db.Close() }() + + timeNow := time.Now() + + emitData(b, ctx, hermesClient, timeNow.Add(-1*time.Hour), 1*time.Hour) + query(b, ctx, db, "1Hour") + + emitData(b, ctx, hermesClient, timeNow.Add(-24*time.Hour), 23*time.Hour) + query(b, ctx, db, "1Day") +} + +func startCommand(b *testing.B, ctx context.Context, dir, name string, args ...string) { + cmd := exec.CommandContext(ctx, name, args...) + cmd.Dir = dir + if err := cmd.Start(); err != nil { + b.Fatalf("Command start failed [%s %v]: %v", name, args, err) + } + + // Stop littering hermes backends on my comupter + b.Cleanup(func() { + if cmd.Process != nil { + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + } + }) +} + +func runCommand(b *testing.B, ctx context.Context, dir, name string, args ...string) { + cmd := exec.CommandContext(ctx, name, args...) + cmd.Dir = dir + if err := cmd.Run(); err != nil { + b.Fatalf("Command run failed [%s %v]: %v", name, args, err) + } +} + +func waitPort(b *testing.B, target string) { + for range 10 { + if conn, err := net.DialTimeout("tcp", target, 500*time.Millisecond); err == nil { + _ = conn.Close() + return + } + time.Sleep(1 * time.Second) + } + b.Fatalf("Network connection timed out: %s\n", target) +} + +func startHermesBackend(b *testing.B, ctx context.Context) *grpc.ClientConn { + b.Log("Starting Hermes backend") + + runCommand(b, ctx, "../../..", "make", "out/backend") + startCommand(b, ctx, "../../..", "./out/backend", "--bind-type", "tcp", "--bind", hermesGrpcConnStr) + waitPort(b, hermesGrpcConnStr) + + hermesConn, err := grpc.NewClient(hermesGrpcConnStr, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + b.Fatalf("Hermes backend gRPC connection failed: %v", err) + } + b.Log("Hermes backend started") + return hermesConn +} + +func startTimescaleGrafana(b *testing.B, ctx context.Context, hermesClient hermesGrpc.ApiClient) *sql.DB { + b.Log("Starting TimescaleDB Grafana Docker") + + runCommand(b, ctx, "../..", "docker", "compose", "up", "-d") + b.Cleanup(func() { + runCommand(b, context.Background(), "../..", "docker", "compose", "down", "-v") + }) + waitPort(b, "localhost:5432") + + b.Log("TimescaleDB Grafana Docker Started") + b.Log("Connecting to TimescaleDB") + + db, err := sql.Open("postgres", timescaleConnStr) + if err != nil { + b.Fatalf("Failed to open database pool: %v", err) + } + for range 10 { + if err := db.PingContext(ctx); err == nil { + break + } + time.Sleep(1 * time.Second) + } + + b.Log("Connected to TimescaleDB") + b.Log("Hermes connecting to TimescaleDB") + + timescaleConfig := map[string]interface{}{ + "host": "localhost", + "port": 5432, + "user": "postgres", + "password": "password", + "database": "hermes", + "sslmode": "disable", + } + + timescaleConfigBytes, err := json.Marshal(timescaleConfig) + if err != nil { + b.Fatalf("Failed to marshal database config map: %v", err) + } + + timescaleProfile := &pb.Profile{ + Name: "TimescaleDB", + Provider: "TimescaleDB", + Settings: string(timescaleConfigBytes), + } + + timescaleProfileID, err := hermesClient.AddProfile(ctx, timescaleProfile) + if err != nil { + b.Fatalf("Hermes failed to add profile: %v", err) + } + + _, err = hermesClient.StartProfile(ctx, timescaleProfileID) + if err != nil { + b.Fatalf("Hermes failed to start profile: %v", err) + } + + b.Log("Hermes connected to TimescaleDB") + return db +} + +func emitData(b *testing.B, ctx context.Context, hermesClient hermesGrpc.ApiClient, timeStart time.Time, timeDuration time.Duration) { + timeSecondsTotal := int(timeDuration.Seconds()) + + sources := []string{"source-alpha", "source-beta"} + + telemetryStream, err := hermesClient.EmitTelemetry(ctx) + if err != nil { + b.Fatalf("Failed to open telemetry stream: %v", err) + } + eventStream, err := hermesClient.EmitEvent(ctx) + if err != nil { + b.Fatalf("Failed to open event stream: %v", err) + } + + for timeSeconds := 0; timeSeconds < timeSecondsTotal; timeSeconds++ { + timeUnix := timeStart.Add(time.Duration(timeSeconds) * time.Second) + source := sources[timeSeconds%len(sources)] + + timeProto := &pb.Time{ + Unix: timestamppb.New(timeUnix), + Sclk: float64(timeSeconds), + } + + telemetryPacket := &pb.SourcedTelemetry{ + Source: source, + Telemetry: &pb.Telemetry{ + Ref: &pb.TelemetryRef{ + Component: "TimescaleDB", + Name: "TestTelemetry1", + }, + Time: timeProto, + Value: &pb.Value{ + Value: &pb.Value_F{ + F: rand.Float64() * 100.0, + }, + }, + Labels: map[string]string{ + "key": "test_key_1", + }, + }, + } + if err := telemetryStream.Send(telemetryPacket); err != nil { + b.Fatalf("Hermes failed to emit telemetry at index %d: %v", timeSeconds, err) + } + + if timeSeconds%60 == 0 { + eventPacket := &pb.SourcedEvent{ + Source: source, + Event: &pb.Event{ + Ref: &pb.EventRef{ + Component: "TimescaleDB", + Name: "TestTelemetry1", + }, + Time: timeProto, + Message: "Test Event Message", + }, + } + if err := eventStream.Send(eventPacket); err != nil { + b.Fatalf("Hermes failed to emit event at index %d: %v", timeSeconds, err) + } + } + + if timeSeconds%(60*60) == 0 { + b.Logf("Emitted %d/%d seconds of data", timeSeconds, timeSecondsTotal) + } + } + + if _, err := telemetryStream.CloseAndRecv(); err != nil { + b.Fatalf("Telemetry stream close acknowledgment failed: %v", err) + } + if _, err := eventStream.CloseAndRecv(); err != nil { + b.Fatalf("Event stream close acknowledgment failed: %v", err) + } +} + +func query(b *testing.B, ctx context.Context, db *sql.DB, name string) { + ds := &Datasource{ + db: db, + } + + timeNow := time.Now() + rawSql := fmt.Sprintf(` +SELECT + time_bucket($__interval, t.ert) AS time_bucket, + d.component, + d.name, + t.source, + t.valueType, + t.key, + AVG(t.integral::double precision) AS val_int, + AVG(t.floating::double precision) AS val_float, + AVG(t.boolval::int::double precision) AS val_bool, + MAX(t.string) AS val_str +FROM telemetryDefs d +JOIN telemetry t ON t.telemetryDefId = d.id +WHERE ((d.component = 'TimescaleDB' AND d.name = 'TestTelemetry1')) + AND ('{}'::text[] = '{}' OR t.source = ANY('{}')) + AND t.ert >= '%s' AND t.ert <= '%s' +GROUP BY time_bucket, d.component, d.name, t.source, t.valueType, t.key +ORDER BY time_bucket ASC;`, timeNow.AddDate(-1, 0, 0).Format(time.RFC3339Nano), timeNow.Format(time.RFC3339Nano)) + + queryModel := queryModel{ + QueryType: "telemetry", + TimeField: "time", + RawSql: &rawSql, + } + queryJSON, err := json.Marshal(queryModel) + if err != nil { + b.Fatalf("Failed to marshal queryModel: %v", err) + } + + request := &backend.QueryDataRequest{ + Queries: []backend.DataQuery{ + { + RefID: "TestQueryId1", + JSON: queryJSON, + Interval: 1 * time.Second, + }, + }, + } + + b.Run(name, func(b *testing.B) { + for i := 0; i < b.N; i++ { + response, err := ds.QueryData(ctx, request) + if err != nil { + b.Fatalf("Failed to query data: %v", err) + } + + if queryResp, exists := response.Responses["TestQueryId1"]; exists { + if queryResp.Status != backend.StatusOK && queryResp.Error != nil { + b.Fatalf("Error returned by query data: %v", queryResp.Error) + } + } else { + b.Fatal("Missing expected data from query data") + } + } + }) +} diff --git a/pkg/grpc/hermes.pb.go b/pkg/grpc/hermes.pb.go index b504939..b8cf8cd 100644 --- a/pkg/grpc/hermes.pb.go +++ b/pkg/grpc/hermes.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: hermes.proto package grpc @@ -27,7 +27,7 @@ var File_hermes_proto protoreflect.FileDescriptor const file_hermes_proto_rawDesc = "" + "\n" + "\fhermes.proto\x1a\x1bgoogle/protobuf/empty.proto\x1a\x10dictionary.proto\x1a\n" + - "file.proto\x1a\tfsw.proto\x1a\tbus.proto\x1a\rprofile.proto\x1a\tmsg.proto2\xbd\x0e\n" + + "file.proto\x1a\tfsw.proto\x1a\tbus.proto\x1a\rprofile.proto\x1a\tmsg.proto2\xbf\x0f\n" + "\x03Api\x12:\n" + "\bSequence\x12\x17.hermes.CommandSequence\x1a\x15.hermes.SequenceReply\x12@\n" + "\vRawSequence\x12\x1a.hermes.RawCommandSequence\x1a\x15.hermes.SequenceReply\x12.\n" + @@ -65,8 +65,10 @@ const file_hermes_proto_rawDesc = "" + ".hermes.Id\x1a\x16.google.protobuf.Empty\x12?\n" + "\rAllDictionary\x12\x16.google.protobuf.Empty\x1a\x16.hermes.DictionaryList\x12G\n" + "\x13SubscribeDictionary\x12\x16.google.protobuf.Empty\x1a\x16.hermes.DictionaryList0\x01\x125\n" + - "\bSubEvent\x12\x11.hermes.BusFilter\x1a\x14.hermes.SourcedEvent0\x01\x12=\n" + - "\fSubTelemetry\x12\x11.hermes.BusFilter\x1a\x18.hermes.SourcedTelemetry0\x01\x12<\n" + + "\bSubEvent\x12\x11.hermes.BusFilter\x1a\x14.hermes.SourcedEvent0\x01\x12;\n" + + "\tEmitEvent\x12\x14.hermes.SourcedEvent\x1a\x16.google.protobuf.Empty(\x01\x12=\n" + + "\fSubTelemetry\x12\x11.hermes.BusFilter\x1a\x18.hermes.SourcedTelemetry0\x01\x12C\n" + + "\rEmitTelemetry\x12\x18.hermes.SourcedTelemetry\x1a\x16.google.protobuf.Empty(\x01\x12<\n" + "\x0fSubFileDownlink\x12\x11.hermes.BusFilter\x1a\x14.hermes.FileDownlink0\x01\x128\n" + "\rSubFileUplink\x12\x11.hermes.BusFilter\x1a\x12.hermes.FileUplink0\x01\x12F\n" + "\x0fSubFileTransfer\x12\x16.google.protobuf.Empty\x1a\x19.hermes.FileTransferState0\x012\x80\x02\n" + @@ -90,10 +92,10 @@ var file_hermes_proto_goTypes = []any{ (*pb.Profile)(nil), // 9: hermes.Profile (*pb.Dictionary)(nil), // 10: hermes.Dictionary (*pb.BusFilter)(nil), // 11: hermes.BusFilter - (*pb.FswConnectionPacket)(nil), // 12: hermes.FswConnectionPacket - (*pb.DownlinkFileChunk)(nil), // 13: hermes.DownlinkFileChunk - (*pb.SourcedEvent)(nil), // 14: hermes.SourcedEvent - (*pb.SourcedTelemetry)(nil), // 15: hermes.SourcedTelemetry + (*pb.SourcedEvent)(nil), // 12: hermes.SourcedEvent + (*pb.SourcedTelemetry)(nil), // 13: hermes.SourcedTelemetry + (*pb.FswConnectionPacket)(nil), // 14: hermes.FswConnectionPacket + (*pb.DownlinkFileChunk)(nil), // 15: hermes.DownlinkFileChunk (*pb.SequenceReply)(nil), // 16: hermes.SequenceReply (*pb.Reply)(nil), // 17: hermes.Reply (*pb.RequestReply)(nil), // 18: hermes.RequestReply @@ -135,51 +137,55 @@ var file_hermes_proto_depIdxs = []int32{ 7, // 24: Api.AllDictionary:input_type -> google.protobuf.Empty 7, // 25: Api.SubscribeDictionary:input_type -> google.protobuf.Empty 11, // 26: Api.SubEvent:input_type -> hermes.BusFilter - 11, // 27: Api.SubTelemetry:input_type -> hermes.BusFilter - 11, // 28: Api.SubFileDownlink:input_type -> hermes.BusFilter - 11, // 29: Api.SubFileUplink:input_type -> hermes.BusFilter - 7, // 30: Api.SubFileTransfer:input_type -> google.protobuf.Empty - 12, // 31: Provider.Connection:input_type -> hermes.FswConnectionPacket - 13, // 32: Provider.File:input_type -> hermes.DownlinkFileChunk - 14, // 33: Provider.Event:input_type -> hermes.SourcedEvent - 15, // 34: Provider.Telemetry:input_type -> hermes.SourcedTelemetry - 16, // 35: Api.Sequence:output_type -> hermes.SequenceReply - 16, // 36: Api.RawSequence:output_type -> hermes.SequenceReply - 17, // 37: Api.Command:output_type -> hermes.Reply - 18, // 38: Api.Request:output_type -> hermes.RequestReply - 17, // 39: Api.RawCommand:output_type -> hermes.Reply - 17, // 40: Api.Uplink:output_type -> hermes.Reply - 19, // 41: Api.GetFsw:output_type -> hermes.Fsw - 20, // 42: Api.AllFsw:output_type -> hermes.FswList - 20, // 43: Api.SubscribeFsw:output_type -> hermes.FswList - 7, // 44: Api.StartProfile:output_type -> google.protobuf.Empty - 7, // 45: Api.StopProfile:output_type -> google.protobuf.Empty - 7, // 46: Api.UpdateProfile:output_type -> google.protobuf.Empty - 6, // 47: Api.AddProfile:output_type -> hermes.Id - 7, // 48: Api.RemoveProfile:output_type -> google.protobuf.Empty - 21, // 49: Api.AllProfiles:output_type -> hermes.ProfileList - 22, // 50: Api.AllProviders:output_type -> hermes.ProfileProviderList - 23, // 51: Api.GetFileTransferState:output_type -> hermes.FileTransferState - 7, // 52: Api.ClearDownlinkTransferState:output_type -> google.protobuf.Empty - 7, // 53: Api.ClearUplinkTransferState:output_type -> google.protobuf.Empty - 22, // 54: Api.SubscribeProviders:output_type -> hermes.ProfileProviderList - 21, // 55: Api.SubscribeProfiles:output_type -> hermes.ProfileList - 10, // 56: Api.GetDictionary:output_type -> hermes.Dictionary - 6, // 57: Api.AddDictionary:output_type -> hermes.Id - 7, // 58: Api.RemoveDictionary:output_type -> google.protobuf.Empty - 24, // 59: Api.AllDictionary:output_type -> hermes.DictionaryList - 24, // 60: Api.SubscribeDictionary:output_type -> hermes.DictionaryList - 14, // 61: Api.SubEvent:output_type -> hermes.SourcedEvent - 15, // 62: Api.SubTelemetry:output_type -> hermes.SourcedTelemetry - 25, // 63: Api.SubFileDownlink:output_type -> hermes.FileDownlink - 26, // 64: Api.SubFileUplink:output_type -> hermes.FileUplink - 23, // 65: Api.SubFileTransfer:output_type -> hermes.FileTransferState - 27, // 66: Provider.Connection:output_type -> hermes.Uplink - 7, // 67: Provider.File:output_type -> google.protobuf.Empty - 7, // 68: Provider.Event:output_type -> google.protobuf.Empty - 7, // 69: Provider.Telemetry:output_type -> google.protobuf.Empty - 35, // [35:70] is the sub-list for method output_type - 0, // [0:35] is the sub-list for method input_type + 12, // 27: Api.EmitEvent:input_type -> hermes.SourcedEvent + 11, // 28: Api.SubTelemetry:input_type -> hermes.BusFilter + 13, // 29: Api.EmitTelemetry:input_type -> hermes.SourcedTelemetry + 11, // 30: Api.SubFileDownlink:input_type -> hermes.BusFilter + 11, // 31: Api.SubFileUplink:input_type -> hermes.BusFilter + 7, // 32: Api.SubFileTransfer:input_type -> google.protobuf.Empty + 14, // 33: Provider.Connection:input_type -> hermes.FswConnectionPacket + 15, // 34: Provider.File:input_type -> hermes.DownlinkFileChunk + 12, // 35: Provider.Event:input_type -> hermes.SourcedEvent + 13, // 36: Provider.Telemetry:input_type -> hermes.SourcedTelemetry + 16, // 37: Api.Sequence:output_type -> hermes.SequenceReply + 16, // 38: Api.RawSequence:output_type -> hermes.SequenceReply + 17, // 39: Api.Command:output_type -> hermes.Reply + 18, // 40: Api.Request:output_type -> hermes.RequestReply + 17, // 41: Api.RawCommand:output_type -> hermes.Reply + 17, // 42: Api.Uplink:output_type -> hermes.Reply + 19, // 43: Api.GetFsw:output_type -> hermes.Fsw + 20, // 44: Api.AllFsw:output_type -> hermes.FswList + 20, // 45: Api.SubscribeFsw:output_type -> hermes.FswList + 7, // 46: Api.StartProfile:output_type -> google.protobuf.Empty + 7, // 47: Api.StopProfile:output_type -> google.protobuf.Empty + 7, // 48: Api.UpdateProfile:output_type -> google.protobuf.Empty + 6, // 49: Api.AddProfile:output_type -> hermes.Id + 7, // 50: Api.RemoveProfile:output_type -> google.protobuf.Empty + 21, // 51: Api.AllProfiles:output_type -> hermes.ProfileList + 22, // 52: Api.AllProviders:output_type -> hermes.ProfileProviderList + 23, // 53: Api.GetFileTransferState:output_type -> hermes.FileTransferState + 7, // 54: Api.ClearDownlinkTransferState:output_type -> google.protobuf.Empty + 7, // 55: Api.ClearUplinkTransferState:output_type -> google.protobuf.Empty + 22, // 56: Api.SubscribeProviders:output_type -> hermes.ProfileProviderList + 21, // 57: Api.SubscribeProfiles:output_type -> hermes.ProfileList + 10, // 58: Api.GetDictionary:output_type -> hermes.Dictionary + 6, // 59: Api.AddDictionary:output_type -> hermes.Id + 7, // 60: Api.RemoveDictionary:output_type -> google.protobuf.Empty + 24, // 61: Api.AllDictionary:output_type -> hermes.DictionaryList + 24, // 62: Api.SubscribeDictionary:output_type -> hermes.DictionaryList + 12, // 63: Api.SubEvent:output_type -> hermes.SourcedEvent + 7, // 64: Api.EmitEvent:output_type -> google.protobuf.Empty + 13, // 65: Api.SubTelemetry:output_type -> hermes.SourcedTelemetry + 7, // 66: Api.EmitTelemetry:output_type -> google.protobuf.Empty + 25, // 67: Api.SubFileDownlink:output_type -> hermes.FileDownlink + 26, // 68: Api.SubFileUplink:output_type -> hermes.FileUplink + 23, // 69: Api.SubFileTransfer:output_type -> hermes.FileTransferState + 27, // 70: Provider.Connection:output_type -> hermes.Uplink + 7, // 71: Provider.File:output_type -> google.protobuf.Empty + 7, // 72: Provider.Event:output_type -> google.protobuf.Empty + 7, // 73: Provider.Telemetry:output_type -> google.protobuf.Empty + 37, // [37:74] is the sub-list for method output_type + 0, // [0:37] is the sub-list for method input_type 0, // [0:0] is the sub-list for extension type_name 0, // [0:0] is the sub-list for extension extendee 0, // [0:0] is the sub-list for field type_name diff --git a/pkg/grpc/hermes_grpc.pb.go b/pkg/grpc/hermes_grpc.pb.go index c2a4c48..23f0276 100644 --- a/pkg/grpc/hermes_grpc.pb.go +++ b/pkg/grpc/hermes_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.2 -// - protoc v4.25.2 +// - protoc v7.35.1 // source: hermes.proto package grpc @@ -48,7 +48,9 @@ const ( Api_AllDictionary_FullMethodName = "/Api/AllDictionary" Api_SubscribeDictionary_FullMethodName = "/Api/SubscribeDictionary" Api_SubEvent_FullMethodName = "/Api/SubEvent" + Api_EmitEvent_FullMethodName = "/Api/EmitEvent" Api_SubTelemetry_FullMethodName = "/Api/SubTelemetry" + Api_EmitTelemetry_FullMethodName = "/Api/EmitTelemetry" Api_SubFileDownlink_FullMethodName = "/Api/SubFileDownlink" Api_SubFileUplink_FullMethodName = "/Api/SubFileUplink" Api_SubFileTransfer_FullMethodName = "/Api/SubFileTransfer" @@ -182,8 +184,12 @@ type ApiClient interface { SubscribeDictionary(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pb.DictionaryList], error) // Subscribe to the event message bus SubEvent(ctx context.Context, in *pb.BusFilter, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pb.SourcedEvent], error) + // Emit to the event message bus + EmitEvent(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[pb.SourcedEvent, emptypb.Empty], error) // Subscribe to the telemetry message bus SubTelemetry(ctx context.Context, in *pb.BusFilter, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pb.SourcedTelemetry], error) + // Emit to the telemetry message bus + EmitTelemetry(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[pb.SourcedTelemetry, emptypb.Empty], error) // Subscribe to the file downlink message bus SubFileDownlink(ctx context.Context, in *pb.BusFilter, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pb.FileDownlink], error) // Subscribe to the file downlink message bus @@ -518,9 +524,22 @@ func (c *apiClient) SubEvent(ctx context.Context, in *pb.BusFilter, opts ...grpc // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type Api_SubEventClient = grpc.ServerStreamingClient[pb.SourcedEvent] +func (c *apiClient) EmitEvent(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[pb.SourcedEvent, emptypb.Empty], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[6], Api_EmitEvent_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[pb.SourcedEvent, emptypb.Empty]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Api_EmitEventClient = grpc.ClientStreamingClient[pb.SourcedEvent, emptypb.Empty] + func (c *apiClient) SubTelemetry(ctx context.Context, in *pb.BusFilter, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pb.SourcedTelemetry], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[6], Api_SubTelemetry_FullMethodName, cOpts...) + stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[7], Api_SubTelemetry_FullMethodName, cOpts...) if err != nil { return nil, err } @@ -537,9 +556,22 @@ func (c *apiClient) SubTelemetry(ctx context.Context, in *pb.BusFilter, opts ... // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type Api_SubTelemetryClient = grpc.ServerStreamingClient[pb.SourcedTelemetry] +func (c *apiClient) EmitTelemetry(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[pb.SourcedTelemetry, emptypb.Empty], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[8], Api_EmitTelemetry_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[pb.SourcedTelemetry, emptypb.Empty]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Api_EmitTelemetryClient = grpc.ClientStreamingClient[pb.SourcedTelemetry, emptypb.Empty] + func (c *apiClient) SubFileDownlink(ctx context.Context, in *pb.BusFilter, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pb.FileDownlink], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[7], Api_SubFileDownlink_FullMethodName, cOpts...) + stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[9], Api_SubFileDownlink_FullMethodName, cOpts...) if err != nil { return nil, err } @@ -558,7 +590,7 @@ type Api_SubFileDownlinkClient = grpc.ServerStreamingClient[pb.FileDownlink] func (c *apiClient) SubFileUplink(ctx context.Context, in *pb.BusFilter, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pb.FileUplink], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[8], Api_SubFileUplink_FullMethodName, cOpts...) + stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[10], Api_SubFileUplink_FullMethodName, cOpts...) if err != nil { return nil, err } @@ -577,7 +609,7 @@ type Api_SubFileUplinkClient = grpc.ServerStreamingClient[pb.FileUplink] func (c *apiClient) SubFileTransfer(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (grpc.ServerStreamingClient[pb.FileTransferState], error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) - stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[9], Api_SubFileTransfer_FullMethodName, cOpts...) + stream, err := c.cc.NewStream(ctx, &Api_ServiceDesc.Streams[11], Api_SubFileTransfer_FullMethodName, cOpts...) if err != nil { return nil, err } @@ -722,8 +754,12 @@ type ApiServer interface { SubscribeDictionary(*emptypb.Empty, grpc.ServerStreamingServer[pb.DictionaryList]) error // Subscribe to the event message bus SubEvent(*pb.BusFilter, grpc.ServerStreamingServer[pb.SourcedEvent]) error + // Emit to the event message bus + EmitEvent(grpc.ClientStreamingServer[pb.SourcedEvent, emptypb.Empty]) error // Subscribe to the telemetry message bus SubTelemetry(*pb.BusFilter, grpc.ServerStreamingServer[pb.SourcedTelemetry]) error + // Emit to the telemetry message bus + EmitTelemetry(grpc.ClientStreamingServer[pb.SourcedTelemetry, emptypb.Empty]) error // Subscribe to the file downlink message bus SubFileDownlink(*pb.BusFilter, grpc.ServerStreamingServer[pb.FileDownlink]) error // Subscribe to the file downlink message bus @@ -821,9 +857,15 @@ func (UnimplementedApiServer) SubscribeDictionary(*emptypb.Empty, grpc.ServerStr func (UnimplementedApiServer) SubEvent(*pb.BusFilter, grpc.ServerStreamingServer[pb.SourcedEvent]) error { return status.Error(codes.Unimplemented, "method SubEvent not implemented") } +func (UnimplementedApiServer) EmitEvent(grpc.ClientStreamingServer[pb.SourcedEvent, emptypb.Empty]) error { + return status.Error(codes.Unimplemented, "method EmitEvent not implemented") +} func (UnimplementedApiServer) SubTelemetry(*pb.BusFilter, grpc.ServerStreamingServer[pb.SourcedTelemetry]) error { return status.Error(codes.Unimplemented, "method SubTelemetry not implemented") } +func (UnimplementedApiServer) EmitTelemetry(grpc.ClientStreamingServer[pb.SourcedTelemetry, emptypb.Empty]) error { + return status.Error(codes.Unimplemented, "method EmitTelemetry not implemented") +} func (UnimplementedApiServer) SubFileDownlink(*pb.BusFilter, grpc.ServerStreamingServer[pb.FileDownlink]) error { return status.Error(codes.Unimplemented, "method SubFileDownlink not implemented") } @@ -1294,6 +1336,13 @@ func _Api_SubEvent_Handler(srv interface{}, stream grpc.ServerStream) error { // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type Api_SubEventServer = grpc.ServerStreamingServer[pb.SourcedEvent] +func _Api_EmitEvent_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(ApiServer).EmitEvent(&grpc.GenericServerStream[pb.SourcedEvent, emptypb.Empty]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Api_EmitEventServer = grpc.ClientStreamingServer[pb.SourcedEvent, emptypb.Empty] + func _Api_SubTelemetry_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(pb.BusFilter) if err := stream.RecvMsg(m); err != nil { @@ -1305,6 +1354,13 @@ func _Api_SubTelemetry_Handler(srv interface{}, stream grpc.ServerStream) error // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. type Api_SubTelemetryServer = grpc.ServerStreamingServer[pb.SourcedTelemetry] +func _Api_EmitTelemetry_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(ApiServer).EmitTelemetry(&grpc.GenericServerStream[pb.SourcedTelemetry, emptypb.Empty]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Api_EmitTelemetryServer = grpc.ClientStreamingServer[pb.SourcedTelemetry, emptypb.Empty] + func _Api_SubFileDownlink_Handler(srv interface{}, stream grpc.ServerStream) error { m := new(pb.BusFilter) if err := stream.RecvMsg(m); err != nil { @@ -1461,11 +1517,21 @@ var Api_ServiceDesc = grpc.ServiceDesc{ Handler: _Api_SubEvent_Handler, ServerStreams: true, }, + { + StreamName: "EmitEvent", + Handler: _Api_EmitEvent_Handler, + ClientStreams: true, + }, { StreamName: "SubTelemetry", Handler: _Api_SubTelemetry_Handler, ServerStreams: true, }, + { + StreamName: "EmitTelemetry", + Handler: _Api_EmitTelemetry_Handler, + ClientStreams: true, + }, { StreamName: "SubFileDownlink", Handler: _Api_SubFileDownlink_Handler, diff --git a/pkg/pb/bus.pb.go b/pkg/pb/bus.pb.go index 53953fd..4522b2c 100644 --- a/pkg/pb/bus.pb.go +++ b/pkg/pb/bus.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: bus.proto package pb diff --git a/pkg/pb/dictionary.pb.go b/pkg/pb/dictionary.pb.go index c6317cf..4e021fc 100644 --- a/pkg/pb/dictionary.pb.go +++ b/pkg/pb/dictionary.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: dictionary.proto package pb diff --git a/pkg/pb/file.pb.go b/pkg/pb/file.pb.go index 66e1aad..34f18be 100644 --- a/pkg/pb/file.pb.go +++ b/pkg/pb/file.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: file.proto package pb diff --git a/pkg/pb/fsw.pb.go b/pkg/pb/fsw.pb.go index 30f6f88..0bc02ef 100644 --- a/pkg/pb/fsw.pb.go +++ b/pkg/pb/fsw.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: fsw.proto package pb diff --git a/pkg/pb/msg.pb.go b/pkg/pb/msg.pb.go index 2946054..18e4773 100644 --- a/pkg/pb/msg.pb.go +++ b/pkg/pb/msg.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: msg.proto package pb diff --git a/pkg/pb/profile.pb.go b/pkg/pb/profile.pb.go index 1623eea..a7d6f07 100644 --- a/pkg/pb/profile.pb.go +++ b/pkg/pb/profile.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: profile.proto package pb diff --git a/pkg/pb/time.pb.go b/pkg/pb/time.pb.go index 9777b76..2bfbeb4 100644 --- a/pkg/pb/time.pb.go +++ b/pkg/pb/time.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: time.proto package pb diff --git a/pkg/pb/type.pb.go b/pkg/pb/type.pb.go index b09a9a6..de6c50d 100644 --- a/pkg/pb/type.pb.go +++ b/pkg/pb/type.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.11 -// protoc v4.25.2 +// protoc v7.35.1 // source: type.proto package pb diff --git a/pkg/rpc/api.go b/pkg/rpc/api.go index de95e29..5379e4d 100644 --- a/pkg/rpc/api.go +++ b/pkg/rpc/api.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "io" "maps" "slices" "strings" @@ -840,6 +841,23 @@ func (r *apiServer) SubEvent(filter *pb.BusFilter, s grpc.ServerStreamingServer[ return nil } +func (r *apiServer) EmitEvent(s grpc.ClientStreamingServer[pb.SourcedEvent, emptypb.Empty]) error { + for { + e, err := s.Recv() + if err == io.EOF { + return s.SendAndClose(&emptypb.Empty{}) + } + if err != nil { + return fmt.Errorf("error whlie receiving event from stream: %w", err) + } + if e == nil || e.Event == nil || e.Event.Ref == nil { + return fmt.Errorf("invalid event packet payload struct") + } + + host.Event.Emit(e) + } +} + // SubTelemetry implements pb.ApiServer. func (r *apiServer) SubTelemetry(filter *pb.BusFilter, s grpc.ServerStreamingServer[pb.SourcedTelemetry]) error { // Validate filter @@ -891,6 +909,23 @@ func (r *apiServer) SubTelemetry(filter *pb.BusFilter, s grpc.ServerStreamingSer return nil } +func (r *apiServer) EmitTelemetry(s grpc.ClientStreamingServer[pb.SourcedTelemetry, emptypb.Empty]) error { + for { + t, err := s.Recv() + if err == io.EOF { + return s.SendAndClose(&emptypb.Empty{}) + } + if err != nil { + return fmt.Errorf("error whlie receiving telemetry from stream: %w", err) + } + if t == nil || t.Telemetry == nil || t.Telemetry.Ref == nil { + return fmt.Errorf("invalid telemetry packet payload struct") + } + + host.Telemetry.Emit(t) + } +} + // SubFileTransfer implements grpc.ApiServer. func (r *apiServer) SubFileTransfer(_ *emptypb.Empty, s grpc.ServerStreamingServer[pb.FileTransferState]) error { host.FileTransfer.On(s.Context(), func(msg *pb.FileTransferState) { diff --git a/proto/grpc/hermes.proto b/proto/grpc/hermes.proto index 4d04695..b76c285 100644 --- a/proto/grpc/hermes.proto +++ b/proto/grpc/hermes.proto @@ -162,9 +162,15 @@ service Api { // Subscribe to the event message bus rpc SubEvent(hermes.BusFilter) returns (stream hermes.SourcedEvent); + // Emit to the event message bus + rpc EmitEvent(stream hermes.SourcedEvent) returns (google.protobuf.Empty); + // Subscribe to the telemetry message bus rpc SubTelemetry(hermes.BusFilter) returns (stream hermes.SourcedTelemetry); + // Emit to the telemetry message bus + rpc EmitTelemetry(stream hermes.SourcedTelemetry) returns (google.protobuf.Empty); + // Subscribe to the file downlink message bus rpc SubFileDownlink(hermes.BusFilter) returns (stream hermes.FileDownlink); diff --git a/src/modules/rpc/src/proto/Api.ts b/src/modules/rpc/src/proto/Api.ts index 4d5e783..6099bb7 100644 --- a/src/modules/rpc/src/proto/Api.ts +++ b/src/modules/rpc/src/proto/Api.ts @@ -110,6 +110,24 @@ export interface ApiClient extends grpc.Client { command(argument: _hermes_CommandValue, options: grpc.CallOptions, callback: grpc.requestCallback<_hermes_Reply__Output>): grpc.ClientUnaryCall; command(argument: _hermes_CommandValue, callback: grpc.requestCallback<_hermes_Reply__Output>): grpc.ClientUnaryCall; + EmitEvent(metadata: grpc.Metadata, options: grpc.CallOptions, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedEvent>; + EmitEvent(metadata: grpc.Metadata, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedEvent>; + EmitEvent(options: grpc.CallOptions, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedEvent>; + EmitEvent(callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedEvent>; + emitEvent(metadata: grpc.Metadata, options: grpc.CallOptions, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedEvent>; + emitEvent(metadata: grpc.Metadata, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedEvent>; + emitEvent(options: grpc.CallOptions, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedEvent>; + emitEvent(callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedEvent>; + + EmitTelemetry(metadata: grpc.Metadata, options: grpc.CallOptions, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedTelemetry>; + EmitTelemetry(metadata: grpc.Metadata, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedTelemetry>; + EmitTelemetry(options: grpc.CallOptions, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedTelemetry>; + EmitTelemetry(callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedTelemetry>; + emitTelemetry(metadata: grpc.Metadata, options: grpc.CallOptions, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedTelemetry>; + emitTelemetry(metadata: grpc.Metadata, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedTelemetry>; + emitTelemetry(options: grpc.CallOptions, callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedTelemetry>; + emitTelemetry(callback: grpc.requestCallback<_google_protobuf_Empty__Output>): grpc.ClientWritableStream<_hermes_SourcedTelemetry>; + GetDictionary(argument: _hermes_Id, metadata: grpc.Metadata, options: grpc.CallOptions, callback: grpc.requestCallback<_hermes_Dictionary__Output>): grpc.ClientUnaryCall; GetDictionary(argument: _hermes_Id, metadata: grpc.Metadata, callback: grpc.requestCallback<_hermes_Dictionary__Output>): grpc.ClientUnaryCall; GetDictionary(argument: _hermes_Id, options: grpc.CallOptions, callback: grpc.requestCallback<_hermes_Dictionary__Output>): grpc.ClientUnaryCall; @@ -293,6 +311,10 @@ export interface ApiHandlers extends grpc.UntypedServiceImplementation { Command: grpc.handleUnaryCall<_hermes_CommandValue__Output, _hermes_Reply>; + EmitEvent: grpc.handleClientStreamingCall<_hermes_SourcedEvent__Output, _google_protobuf_Empty>; + + EmitTelemetry: grpc.handleClientStreamingCall<_hermes_SourcedTelemetry__Output, _google_protobuf_Empty>; + GetDictionary: grpc.handleUnaryCall<_hermes_Id__Output, _hermes_Dictionary>; GetFileTransferState: grpc.handleUnaryCall<_google_protobuf_Empty__Output, _hermes_FileTransferState>; @@ -349,6 +371,8 @@ export interface ApiDefinition extends grpc.ServiceDefinition { ClearDownlinkTransferState: MethodDefinition<_google_protobuf_Empty, _google_protobuf_Empty, _google_protobuf_Empty__Output, _google_protobuf_Empty__Output> ClearUplinkTransferState: MethodDefinition<_google_protobuf_Empty, _google_protobuf_Empty, _google_protobuf_Empty__Output, _google_protobuf_Empty__Output> Command: MethodDefinition<_hermes_CommandValue, _hermes_Reply, _hermes_CommandValue__Output, _hermes_Reply__Output> + EmitEvent: MethodDefinition<_hermes_SourcedEvent, _google_protobuf_Empty, _hermes_SourcedEvent__Output, _google_protobuf_Empty__Output> + EmitTelemetry: MethodDefinition<_hermes_SourcedTelemetry, _google_protobuf_Empty, _hermes_SourcedTelemetry__Output, _google_protobuf_Empty__Output> GetDictionary: MethodDefinition<_hermes_Id, _hermes_Dictionary, _hermes_Id__Output, _hermes_Dictionary__Output> GetFileTransferState: MethodDefinition<_google_protobuf_Empty, _hermes_FileTransferState, _google_protobuf_Empty__Output, _hermes_FileTransferState__Output> GetFsw: MethodDefinition<_hermes_Id, _hermes_Fsw, _hermes_Id__Output, _hermes_Fsw__Output> diff --git a/src/modules/rpc/src/proto/hermes.json b/src/modules/rpc/src/proto/hermes.json index f80c77f..74fbc57 100644 --- a/src/modules/rpc/src/proto/hermes.json +++ b/src/modules/rpc/src/proto/hermes.json @@ -120,11 +120,21 @@ "responseType": "hermes.SourcedEvent", "responseStream": true }, + "EmitEvent": { + "requestType": "hermes.SourcedEvent", + "requestStream": true, + "responseType": "google.protobuf.Empty" + }, "SubTelemetry": { "requestType": "hermes.BusFilter", "responseType": "hermes.SourcedTelemetry", "responseStream": true }, + "EmitTelemetry": { + "requestType": "hermes.SourcedTelemetry", + "requestStream": true, + "responseType": "google.protobuf.Empty" + }, "SubFileDownlink": { "requestType": "hermes.BusFilter", "responseType": "hermes.FileDownlink",