Function Test_Client_QueueUpdate

client_test.go:6554–6693

(t *testing.T)

}

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

	ctx := context.Background()

	type testBundle struct {
		schema string
	}

	setup := func(t *testing.T) (*Client[pgx.Tx], *testBundle) {
		t.Helper()

		var (
			dbPool = riversharedtest.DBPool(ctx, t)
			driver = riverpgxv5.New(dbPool)
			schema = riverdbtest.TestSchema(ctx, t, driver, nil)
			config = newTestConfig(t, schema)
			client = newTestClient(t, dbPool, config)
		)

		return client, &testBundle{
			schema: schema,
		}
	}

	t.Run("UpdatesQueueMetadata", func(t *testing.T) {
		t.Parallel()

		client, bundle := setup(t)
		startClient(ctx, t, client)

		type metadataUpdatePayload struct {
			Action   string          `json:"action"`
			Metadata json.RawMessage `json:"metadata"`
			Queue    string          `json:"queue"`
		}
		type notification struct {
			topic   notifier.NotificationTopic
			payload metadataUpdatePayload
		}
		notifyCh := make(chan notification, 10)

		handleNotification := func(topic notifier.NotificationTopic, payload string) {
			t.Logf("received notification: %s, %q", topic, payload)
			notif := notification{topic: topic}
			require.NoError(t, json.Unmarshal([]byte(payload), &notif.payload))
			notifyCh <- notif
		}

		sub, err := client.notifier.Listen(ctx, notifier.NotificationTopicControl, handleNotification)
		require.NoError(t, err)
		t.Cleanup(func() { sub.Unlisten(ctx) })

		queue := testfactory.Queue(ctx, t, client.driver.GetExecutor(), &testfactory.QueueOpts{Schema: bundle.schema})
		require.Equal(t, []byte(`{}`), queue.Metadata)

		queue, err = client.QueueUpdate(ctx, queue.Name, &QueueUpdateParams{
			Metadata: []byte(`{"foo":"bar"}`),