Function TestPersistentConnection

orchestration/orchestrator_test.go:664–778

TestPersistentConnection makes sure updating the ingress doesn't intefere with existing connections

(t *testing.T)

// TestPersistentConnection makes sure updating the ingress doesn't intefere with existing connections
func TestPersistentConnection(t *testing.T) {
	const (
		hostname = "http://ws.tunnel.org"
	)
	msg := t.Name()
	originDialer := ingress.NewOriginDialer(ingress.OriginConfig{
		DefaultDialer:   testDefaultDialer,
		TCPWriteTimeout: 1 * time.Second,
	}, &testLogger)
	initConfig := &Config{
		Ingress:             &ingress.Ingress{},
		OriginDialerService: originDialer,
	}
	orchestrator, err := NewOrchestrator(t.Context(), initConfig, testTags, []ingress.Rule{}, &testLogger)
	require.NoError(t, err)

	wsOrigin := httptest.NewServer(http.HandlerFunc(wsEcho))
	defer wsOrigin.Close()

	tcpOrigin, err := net.Listen("tcp", "127.0.0.1:0")
	require.NoError(t, err)
	defer tcpOrigin.Close()

	configWithWSAndWarp := []byte(fmt.Sprintf(`
{
    "ingress": [
        {
            "service": "%s"
        }
    ],
    "warp-routing": {
    }
}
`, wsOrigin.URL))

	updateWithValidation(t, orchestrator, 1, configWithWSAndWarp)

	originProxy, err := orchestrator.GetOriginProxy()
	require.NoError(t, err)

	wsReqReader, wsReqWriter := io.Pipe()
	wsRespReadWriter := newRespReadWriteFlusher()

	tcpReqReader, tcpReqWriter := io.Pipe()
	tcpRespReadWriter := newRespReadWriteFlusher()

	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()

	var wg sync.WaitGroup
	wg.Add(3)
	// Start TCP origin
	go func() {
		defer wg.Done()
		conn, err := tcpOrigin.Accept()
		assert.NoError(t, err)
		defer conn.Close()