Function launchConnWithStreams

middleware/grpc_stats_test.go:398–440

(t *testing.T, ctx context.Context, listener net.Listener, streamsCount int)

}

func launchConnWithStreams(t *testing.T, ctx context.Context, listener net.Listener, streamsCount int) *grpc.ClientConn {
	t.Helper()

	conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(10e6), grpc.MaxCallSendMsgSize(10e6)))
	require.NoError(t, err)

	fc := middleware_test.NewEchoServerClient(conn)

	streams := make([]middleware_test.EchoServer_ProcessClient, streamsCount)
	for i := 0; i < streamsCount; i++ {
		stream, err := fc.Process(context.Background())
		require.NoError(t, err)
		streams[i] = stream
	}

	// Keep streams alive by continuously sending messages
	for i := 0; i < len(streams); i++ {
		go func(streamIndex int) {
			for {
				select {
				case <-ctx.Done():
					if err := streams[streamIndex].CloseSend(); err != nil {
						t.Log("Error closing stream", err)
					}
					return
				default:
					msg := &middleware_test.Msg{
						Body: []byte(generateString(100)), // Small message to keep stream alive
					}
					if err := streams[streamIndex].Send(msg); err != nil {
						return
					}
					if _, err := streams[streamIndex].Recv(); err != nil {
						return
					}
					time.Sleep(10 * time.Millisecond)
				}
			}
		}(i)
	}

	return conn
}

type halfEcho struct {
	log func(args ...interface{})