Function TestTCPTransportWriterAcquireTimeout

kv/memberlist/tcp_transport_test.go:124–156

(t *testing.T)

}

func TestTCPTransportWriterAcquireTimeout(t *testing.T) {
	// Listen for TCP connections on a random port
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	require.NoError(t, err)
	defer listener.Close()

	logs := &concurrency.SyncBuffer{}
	logger := log.NewLogfmtLogger(logs)

	cfg := TCPTransportConfig{}
	flagext.DefaultValues(&cfg)
	cfg.BindAddrs = getLocalhostAddrs()
	cfg.MaxConcurrentWrites = 1
	cfg.AcquireWriterTimeout = 1 * time.Millisecond // very short timeout
	transport, err := NewTCPTransport(cfg, logger, nil)
	require.NoError(t, err)

	writeCt := 100
	var reqWg sync.WaitGroup
	for i := 0; i < writeCt; i++ {
		reqWg.Add(1)
		go func() {
			defer reqWg.Done()
			transport.WriteTo([]byte("test"), listener.Addr().String()) // nolint:errcheck
		}()
	}
	reqWg.Wait()

	require.NoError(t, transport.Shutdown())
	gotErrorCt := strings.Count(logs.String(), "WriteTo failed to acquire a writer. Dropping message")
	assert.Less(t, gotErrorCt, writeCt, "expected to have less errors (%d) than total writes (%d). Some writes should pass.", gotErrorCt, writeCt)
	assert.NotZero(t, gotErrorCt, "expected errors, got none")
}

func TestFinalAdvertiseAddr(t *testing.T) {
	tests := map[string]struct {