Method SpiteStream

internal/bridge/mockserver_test.go:135–196

-- SpiteStream (bidirectional) ---------------------------------------------

(stream listenerrpc.ListenerRPC_SpiteStreamServer)

// -- SpiteStream (bidirectional) ---------------------------------------------

func (s *testServer) SpiteStream(stream listenerrpc.ListenerRPC_SpiteStreamServer) error {
	// Validate pipeline_id metadata (mirrors real server).
	md, ok := metadata.FromIncomingContext(stream.Context())
	if !ok {
		return status.Error(codes.InvalidArgument, "missing metadata")
	}
	pids := md.Get("pipeline_id")
	if len(pids) == 0 || pids[0] == "" {
		return status.Error(codes.InvalidArgument, "missing pipeline_id")
	}
	s.mu.Lock()
	missingPipeline := s.missingPipeline
	s.mu.Unlock()
	if missingPipeline {
		return status.Error(codes.NotFound, "Pipeline not found")
	}

	ctx := stream.Context()
	errCh := make(chan error, 2)

	// Send requests to bridge.
	go func() {
		for {
			select {
			case req, ok := <-s.spiteReqCh:
				if !ok {
					errCh <- context.Canceled // signal stream end
					return
				}
				if err := stream.Send(req); err != nil {
					errCh <- err
					return
				}
			case <-ctx.Done():
				return
			}
		}
	}()

	// Receive responses from bridge.
	go func() {
		for {
			resp, err := stream.Recv()
			if err != nil {
				errCh <- err
				return
			}
			select {
			case s.spiteRespCh <- resp:
			default:
				// drop if buffer full
			}
		}
	}()

	select {
	case <-ctx.Done():
		return ctx.Err()