Function TestPipelineFlushForSingleRequests

pgconn/pgconn_test.go:3627–3724

(t *testing.T)

}

func TestPipelineFlushForSingleRequests(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
	defer cancel()

	pgConn, err := pgconn.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
	require.NoError(t, err)
	defer closeConn(t, pgConn)

	pipeline := pgConn.StartPipeline(ctx)

	pipeline.SendPrepare("ps", "select $1::text as msg", nil)
	pipeline.SendFlushRequest()
	err = pipeline.Flush()
	require.NoError(t, err)

	results, err := pipeline.GetResults()
	require.NoError(t, err)
	sd, ok := results.(*pgconn.StatementDescription)
	require.Truef(t, ok, "expected StatementDescription, got: %#v", results)
	require.Len(t, sd.Fields, 1)
	require.Equal(t, "msg", sd.Fields[0].Name)
	require.Equal(t, []uint32{pgtype.TextOID}, sd.ParamOIDs)

	results, err = pipeline.GetResults()
	require.NoError(t, err)
	require.Nil(t, results)

	pipeline.SendQueryPrepared(`ps`, [][]byte{[]byte("hello")}, nil, nil)
	pipeline.SendFlushRequest()
	err = pipeline.Flush()
	require.NoError(t, err)

	results, err = pipeline.GetResults()
	require.NoError(t, err)
	rr, ok := results.(*pgconn.ResultReader)
	require.Truef(t, ok, "expected ResultReader, got: %#v", results)
	readResult := rr.Read()
	require.NoError(t, readResult.Err)
	require.Len(t, readResult.Rows, 1)
	require.Len(t, readResult.Rows[0], 1)
	require.Equal(t, "hello", string(readResult.Rows[0][0]))

	results, err = pipeline.GetResults()
	require.NoError(t, err)
	require.Nil(t, results)

	pipeline.SendDeallocate("ps")
	pipeline.SendFlushRequest()
	err = pipeline.Flush()
	require.NoError(t, err)

	results, err = pipeline.GetResults()
	require.NoError(t, err)
	_, ok = results.(*pgconn.CloseComplete)
	require.Truef(t, ok, "expected CloseComplete, got: %#v", results)