Method JobStream

internal/bridge/mockserver_test.go:200–253

-- JobStream (bidirectional) -----------------------------------------------

(stream listenerrpc.ListenerRPC_JobStreamServer)

// -- JobStream (bidirectional) -----------------------------------------------

func (s *testServer) JobStream(stream listenerrpc.ListenerRPC_JobStreamServer) error {
	s.mu.Lock()
	missingListener := s.missingListener
	disconnectCh := s.jobStreamDisconnectCh
	s.mu.Unlock()
	if missingListener {
		return status.Error(codes.NotFound, "Listener not found")
	}

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

	// Send job control messages to bridge.
	go func() {
		for {
			select {
			case ctrl, ok := <-s.jobCtrlCh:
				if !ok {
					return
				}
				if err := stream.Send(ctrl); err != nil {
					errCh <- err
					return
				}
			case <-ctx.Done():
				return
			}
		}
	}()

	// Receive job status from bridge.
	go func() {
		for {
			st, err := stream.Recv()
			if err != nil {
				errCh <- err
				return
			}
			select {
			case s.jobStatusCh <- st:
			default:
			}
		}
	}()

	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-disconnectCh:
		return status.Error(codes.Unavailable, "job stream disconnected")
	case err := <-errCh:
		return err
	}
}

// -- Helpers for reading captured state --------------------------------------