Function Example_queuePause

example_queue_pause_test.go:36–111

Example_queuePause demonstrates how to pause queues to prevent them from working new jobs, and later resume them.

()

// Example_queuePause demonstrates how to pause queues to prevent them from
// working new jobs, and later resume them.
func Example_queuePause() {
	ctx := context.Background()

	dbPool, err := pgxpool.New(ctx, riversharedtest.TestDatabaseURL())
	if err != nil {
		panic(err)
	}
	defer dbPool.Close()

	const (
		unreliableQueue = "unreliable_external_service"
		reliableQueue   = "reliable_jobs"
	)

	workers := river.NewWorkers()
	jobWorkedCh := make(chan string)
	river.AddWorker(workers, &ReportingWorker{jobWorkedCh: jobWorkedCh})

	riverClient, err := river.NewClient(riverpgxv5.New(dbPool), initTestConfig(ctx, dbPool, &river.Config{
		Queues: map[string]river.QueueConfig{
			unreliableQueue: {MaxWorkers: 10},
			reliableQueue:   {MaxWorkers: 10},
		},
		Workers: workers,
	}))
	if err != nil {
		panic(err)
	}

	if err := riverClient.Start(ctx); err != nil {
		panic(err)
	}

	// Out of example scope, but used to wait until a queue is paused or unpaused.
	subscribeChan, subscribeCancel := riverClient.Subscribe(river.EventKindQueuePaused, river.EventKindQueueResumed)
	defer subscribeCancel()

	fmt.Printf("Pausing %s queue\n", unreliableQueue)
	if err := riverClient.QueuePause(ctx, unreliableQueue, nil); err != nil {
		panic(err)
	}

	// Wait for queue to be paused:
	waitOrTimeout(subscribeChan)

	fmt.Println("Inserting one job each into unreliable and reliable queues")
	if _, err = riverClient.Insert(ctx, ReportingArgs{}, &river.InsertOpts{Queue: unreliableQueue}); err != nil {
		panic(err)
	}
	if _, err = riverClient.Insert(ctx, ReportingArgs{}, &river.InsertOpts{Queue: reliableQueue}); err != nil {
		panic(err)
	}
	// The unreliable queue is paused so its job should get worked yet, while
	// reliable queue is not paused so its job should get worked immediately:
	receivedQueue := waitOrTimeout(jobWorkedCh)
	fmt.Printf("Job worked on %s queue\n", receivedQueue)

	// Resume the unreliable queue so it can work the job: