Function TestConnectSession

proxycore/session_test.go:33–114

(t *testing.T)

)

func TestConnectSession(t *testing.T) {
	logger, _ := zap.NewDevelopment()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	const supported = primitive.ProtocolVersion4

	c := NewMockCluster(net.ParseIP("127.0.0.0"), 9042)

	err := c.Add(ctx, 1)
	require.NoError(t, err)

	err = c.Add(ctx, 2)
	require.NoError(t, err)

	err = c.Add(ctx, 3)
	require.NoError(t, err)

	cluster, err := ConnectCluster(ctx, ClusterConfig{
		Version:           supported,
		Resolver:          NewResolver("127.0.0.1:9042"),
		ReconnectPolicy:   NewReconnectPolicyWithDelays(200*time.Millisecond, time.Second),
		RefreshWindow:     100 * time.Millisecond,
		ConnectTimeout:    10 * time.Second,
		Logger:            logger,
		HeartBeatInterval: 30 * time.Second,
		IdleTimeout:       60 * time.Second,
	})
	require.NoError(t, err)

	session, err := ConnectSession(ctx, cluster, SessionConfig{
		ReconnectPolicy:   NewReconnectPolicyWithDelays(200*time.Millisecond, time.Second),
		NumConns:          2,
		Version:           supported,
		ConnectTimeout:    10 * time.Second,
		HeartBeatInterval: 30 * time.Second,
		IdleTimeout:       60 * time.Second,
	})
	require.NoError(t, err)

	newHost := func(addr string) *Host {
		return &Host{Endpoint: &defaultEndpoint{addr: addr}}
	}

	var wg sync.WaitGroup

	wg.Add(3)

	err = session.Send(newHost("127.0.0.1:9042"), &testSessionRequest{t: t, rpcAddr: "127.0.0.1", wg: &wg})
	require.NoError(t, err)

	err = session.Send(newHost("127.0.0.2:9042"), &testSessionRequest{t: t, rpcAddr: "127.0.0.2", wg: &wg})
	require.NoError(t, err)

	err = session.Send(newHost("127.0.0.3:9042"), &testSessionRequest{t: t, rpcAddr: "127.0.0.3", wg: &wg})
	require.NoError(t, err)