phill/tessera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor stream management: rename stream methods to use 'Client' prefix for clarity and update related logic in gRPC and manager files

Browse Source
This commit is contained in:
Phillip
2025-08-21 12:53:20 +00:00
parent 2484a33945
commit 6ad3deb5fc
7 changed files with 45 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
services/data_service/cmd/stream_tap/main.go
View File

@@ -92,7 +92,7 @@ func main() {
flag.Parse() flag.Parse()
if len(ids) == 0 { if len(ids) == 0 {
fmt.Fprintln(os.Stderr, "provide at least one --id (provider:subject or canonical key)") _, _ = fmt.Fprintln(os.Stderr, "provide at least one --id (provider:subject or canonical key)")
os.Exit(2) os.Exit(2)
} }
@@ -104,16 +104,24 @@ func main() {
grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithTransportCredentials(insecure.NewCredentials()),
) )
if err != nil { if err != nil {
fmt.Fprintf(os.Stderr, "new control client: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "new control client: %v\n", err)
os.Exit(1) os.Exit(1)
} }
defer ccCtl.Close() defer func(ccCtl *grpc.ClientConn) {
err := ccCtl.Close()
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "close control client: %v\n", err)
os.Exit(1)
} else {
fmt.Println("closed control client")
}
}(ccCtl)
ccCtl.Connect() ccCtl.Connect()
ctlConnCtx, cancelCtlConn := context.WithTimeout(ctx, timeout) ctlConnCtx, cancelCtlConn := context.WithTimeout(ctx, timeout)
if err := waitReady(ctlConnCtx, ccCtl); err != nil { if err := waitReady(ctlConnCtx, ccCtl); err != nil {
cancelCtlConn() cancelCtlConn()
fmt.Fprintf(os.Stderr, "connect control: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "connect control: %v\n", err)
os.Exit(1) os.Exit(1)
} }
cancelCtlConn() cancelCtlConn()
@@ -124,7 +132,7 @@ func main() {
startResp, err := ctl.StartStream(ctxStart, &pb.StartStreamRequest{}) startResp, err := ctl.StartStream(ctxStart, &pb.StartStreamRequest{})
cancelStart() cancelStart()
if err != nil { if err != nil {
fmt.Fprintf(os.Stderr, "StartStream: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "StartClientStream: %v\n", err)
os.Exit(1) os.Exit(1)
} }
streamUUID := startResp.GetStreamUuid() streamUUID := startResp.GetStreamUuid()
@@ -134,7 +142,7 @@ func main() {
for _, s := range ids { for _, s := range ids {
key, err := toIdentifierKey(s) key, err := toIdentifierKey(s)
if err != nil { if err != nil {
fmt.Fprintf(os.Stderr, "bad --id: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "bad --id: %v\n", err)
os.Exit(2) os.Exit(2)
} }
pbIDs = append(pbIDs, &pb.Identifier{Key: key}) pbIDs = append(pbIDs, &pb.Identifier{Key: key})
@@ -147,7 +155,7 @@ func main() {
}) })
cancelCfg() cancelCfg()
if err != nil { if err != nil {
fmt.Fprintf(os.Stderr, "ConfigureStream: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "ConfigureClientStream: %v\n", err)
os.Exit(1) os.Exit(1)
} }
fmt.Printf("configured %d identifiers\n", len(pbIDs)) fmt.Printf("configured %d identifiers\n", len(pbIDs))
@@ -157,16 +165,24 @@ func main() {
grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithTransportCredentials(insecure.NewCredentials()),
) )
if err != nil { if err != nil {
fmt.Fprintf(os.Stderr, "new streaming client: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "new streaming client: %v\n", err)
os.Exit(1) os.Exit(1)
} }
defer ccStr.Close() defer func(ccStr *grpc.ClientConn) {
err := ccStr.Close()
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "close streaming client: %v\n", err)
os.Exit(1)
} else {
fmt.Println("closed streaming client")
}
}(ccStr)
ccStr.Connect() ccStr.Connect()
strConnCtx, cancelStrConn := context.WithTimeout(ctx, timeout) strConnCtx, cancelStrConn := context.WithTimeout(ctx, timeout)
if err := waitReady(strConnCtx, ccStr); err != nil { if err := waitReady(strConnCtx, ccStr); err != nil {
cancelStrConn() cancelStrConn()
fmt.Fprintf(os.Stderr, "connect streaming: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "connect streaming: %v\n", err)
os.Exit(1) os.Exit(1)
} }
cancelStrConn() cancelStrConn()
@@ -178,7 +194,7 @@ func main() {
stream, err := str.ConnectStream(streamCtx, &pb.ConnectStreamRequest{StreamUuid: streamUUID}) stream, err := str.ConnectStream(streamCtx, &pb.ConnectStreamRequest{StreamUuid: streamUUID})
if err != nil { if err != nil {
fmt.Fprintf(os.Stderr, "ConnectStream: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "ConnectClientStream: %v\n", err)
os.Exit(1) os.Exit(1)
} }
fmt.Println("connected; streaming… (Ctrl-C to quit)") fmt.Println("connected; streaming… (Ctrl-C to quit)")
@@ -194,7 +210,7 @@ func main() {
if ctx.Err() != nil { if ctx.Err() != nil {
return return
} }
fmt.Fprintf(os.Stderr, "recv: %v\n", err) _, _ = fmt.Fprintf(os.Stderr, "recv: %v\n", err)
os.Exit(1) os.Exit(1)
} }
id := msg.GetIdentifier() id := msg.GetIdentifier()

16
services/data_service/internal/manager/manager.go
View File

@@ -39,7 +39,7 @@ func NewManager(router *router.Router) *Manager {
} }
} }
func (m *Manager) StartStream() (uuid.UUID, error) { func (m *Manager) StartClientStream() (uuid.UUID, error) {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
@@ -50,7 +50,7 @@ func (m *Manager) StartStream() (uuid.UUID, error) {
OutChannel: nil, OutChannel: nil,
Timer: time.AfterFunc(1*time.Minute, func() { Timer: time.AfterFunc(1*time.Minute, func() {
fmt.Printf("stream %s expired due to inactivity\n", streamID) fmt.Printf("stream %s expired due to inactivity\n", streamID)
err := m.StopStream(streamID) err := m.StopClientStream(streamID)
if err != nil { if err != nil {
fmt.Printf("failed to stop stream after timeout: %v\n", err) fmt.Printf("failed to stop stream after timeout: %v\n", err)
} }
@@ -60,7 +60,7 @@ func (m *Manager) StartStream() (uuid.UUID, error) {
return streamID, nil return streamID, nil
} }
func (m *Manager) ConfigureStream(streamID uuid.UUID, newIds []domain.Identifier) error { func (m *Manager) ConfigureClientStream(streamID uuid.UUID, newIds []domain.Identifier) error {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
@@ -150,8 +150,8 @@ func (m *Manager) ConfigureStream(streamID uuid.UUID, newIds []domain.Identifier
return nil return nil
} }
func (m *Manager) StopStream(streamID uuid.UUID) error { func (m *Manager) StopClientStream(streamID uuid.UUID) error {
m.DisconnectStream(streamID) m.DisconnectClientStream(streamID)
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
@@ -186,7 +186,7 @@ func (m *Manager) StopStream(streamID uuid.UUID) error {
return nil return nil
} }
func (m *Manager) ConnectStream(streamID uuid.UUID) (<-chan domain.Message, error) { func (m *Manager) ConnectClientStream(streamID uuid.UUID) (<-chan domain.Message, error) {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
@@ -214,7 +214,7 @@ func (m *Manager) ConnectStream(streamID uuid.UUID) (<-chan domain.Message, erro
return ch, nil return ch, nil
} }
func (m *Manager) DisconnectStream(streamID uuid.UUID) { func (m *Manager) DisconnectClientStream(streamID uuid.UUID) {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
@@ -232,7 +232,7 @@ func (m *Manager) DisconnectStream(streamID uuid.UUID) {
stream.Timer = time.AfterFunc(1*time.Minute, func() { stream.Timer = time.AfterFunc(1*time.Minute, func() {
fmt.Printf("stream %s expired due to inactivity\n", streamID) fmt.Printf("stream %s expired due to inactivity\n", streamID)
err := m.StopStream(streamID) err := m.StopClientStream(streamID)
if err != nil { if err != nil {
fmt.Printf("failed to stop stream after disconnect: %v\n", err) fmt.Printf("failed to stop stream after disconnect: %v\n", err)
} }

6
services/data_service/internal/server/gprc_control_server.go
View File

@@ -22,7 +22,7 @@ func NewGRPCControlServer(m *manager.Manager) *GRPCControlServer {
} }
func (s *GRPCControlServer) StartStream(_ context.Context, _ *pb.StartStreamRequest) (*pb.StartStreamResponse, error) { func (s *GRPCControlServer) StartStream(_ context.Context, _ *pb.StartStreamRequest) (*pb.StartStreamResponse, error) {
streamID, err := s.manager.StartStream() streamID, err := s.manager.StartClientStream()
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to start stream: %w", err) return nil, fmt.Errorf("failed to start stream: %w", err)
} }
@@ -44,7 +44,7 @@ func (s *GRPCControlServer) ConfigureStream(_ context.Context, req *pb.Configure
ids = append(ids, id) ids = append(ids, id)
} }
if err := s.manager.ConfigureStream(streamID, ids); err != nil { if err := s.manager.ConfigureClientStream(streamID, ids); err != nil {
return nil, status.Errorf(codes.InvalidArgument, "configure failed: %v", err) return nil, status.Errorf(codes.InvalidArgument, "configure failed: %v", err)
} }
return &pb.ConfigureStreamResponse{}, nil return &pb.ConfigureStreamResponse{}, nil
@@ -55,7 +55,7 @@ func (s *GRPCControlServer) StopStream(_ context.Context, req *pb.StopStreamRequ
if err != nil { if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "invalid stream_uuid %q: %v", req.StreamUuid, err) return nil, status.Errorf(codes.InvalidArgument, "invalid stream_uuid %q: %v", req.StreamUuid, err)
} }
if err := s.manager.StopStream(streamID); err != nil { if err := s.manager.StopClientStream(streamID); err != nil {
return nil, status.Errorf(codes.Internal, "failed to stop stream: %v", err) return nil, status.Errorf(codes.Internal, "failed to stop stream: %v", err)
} }
return &pb.StopStreamResponse{}, nil return &pb.StopStreamResponse{}, nil

4
services/data_service/internal/server/grpc_streaming_server.go
View File

@@ -23,7 +23,7 @@ func (s *GRPCStreamingServer) ConnectStream(req *pb.ConnectStreamRequest, stream
return fmt.Errorf("invalid UUID: %w", err) return fmt.Errorf("invalid UUID: %w", err)
} }
ch, err := s.manager.ConnectStream(streamUUID) ch, err := s.manager.ConnectClientStream(streamUUID)
if err != nil { if err != nil {
return fmt.Errorf("failed to connect: %w", err) return fmt.Errorf("failed to connect: %w", err)
} }
@@ -31,7 +31,7 @@ func (s *GRPCStreamingServer) ConnectStream(req *pb.ConnectStreamRequest, stream
for { for {
select { select {
case <-stream.Context().Done(): case <-stream.Context().Done():
s.manager.DisconnectStream(streamUUID) s.manager.DisconnectClientStream(streamUUID)
return nil return nil
case msg, ok := <-ch: case msg, ok := <-ch:
if !ok { if !ok {

4
services/data_service/internal/server/socket_streaming_server.go
View File

@@ -66,12 +66,12 @@ func (s *SocketStreamingServer) handleConnection(conn net.Conn) {
return return
} }
outCh, err := s.manager.ConnectStream(streamUUID) outCh, err := s.manager.ConnectClientStream(streamUUID)
if err != nil { if err != nil {
_, _ = fmt.Fprintf(conn, "Failed to connect to stream: %v\n", err) _, _ = fmt.Fprintf(conn, "Failed to connect to stream: %v\n", err)
return return
} }
defer s.manager.DisconnectStream(streamUUID) defer s.manager.DisconnectClientStream(streamUUID)
writer := bufio.NewWriterSize(conn, 256*1024) writer := bufio.NewWriterSize(conn, 256*1024)
defer func(w *bufio.Writer) { defer func(w *bufio.Writer) {

1
services/data_service/internal/worker/stateful.go Normal file
View File

@@ -0,0 +1 @@
package worker

1
services/data_service/internal/worker/stateless.go Normal file
View File

@@ -0,0 +1 @@
package worker