notifysemaphore

Documentation

Overview

NotifySemaphore is a utility library for consumers using LISTEN / NOTIFY to avoid polling the database for new work.

Usage

NotifySemaphore supports multiple concurrent channels, but it does not support concurrent access to the same notification channel. An attempt to do so might result in undefined behaviour or panics. If you need support for multiple readers on the same channel, you should look at other solutions, such as NotifyDispatcher.

An example of the intended usage pattern:

package main

import (
    "github.com/lib/pq"
    "github.com/johto/notifyutils/notifysemaphore"
    "database/sql"
    "time"
)

func work() {
    // Fetch and process work from the database.  It is crucial to process
    // *all* available work, not just one task.
    for {
        task := getWorkFromDatabase()
        if task == nil {
            return
        }

        go doWorkOnTask(task)
    }
}

func main() {
    listener := pq.NewListener("", 15 * time.Second, time.Minute, nil)
    notifysemaphore := notifysemaphore.NewNotifySemaphore(listener)

    // It is important here that the order of operations is:
    //   1) Listen()
    //   2) Process *all* work
    //   3) Wait for a notification (possibly queued while in step 2)
    //   4) Go to 2
    //
    // Following this order guarantees that there will never be work
    // available in the database for extended periods of time without your
    // application knowing about it.
    sem, err := notifysemaphore.Listen("getwork")
    if err != nil {
        panic(err)
    }

    for {
        work()
        <-sem
    }
}

Index

• type NotifySemaphore
  • func NewNotifySemaphore(listener *pq.Listener) *NotifySemaphore
  • func (s *NotifySemaphore) Close() error
  • func (s *NotifySemaphore) Listen(channel string) (<-chan *pq.Notification, error)
  • func (s *NotifySemaphore) Ping() error
  • func (s *NotifySemaphore) SetBroadcastOnPingTimeout(broadcastOnPingTimeout bool)
  • func (s *NotifySemaphore) SetPingInterval(interval time.Duration)
  • func (s *NotifySemaphore) Unlisten(channel string) error

Types

type NotifySemaphore

type NotifySemaphore struct {
	// contains filtered or unexported fields
}

func NewNotifySemaphore

func NewNotifySemaphore(listener *pq.Listener) *NotifySemaphore

func (*NotifySemaphore) Close

func (s *NotifySemaphore) Close() error

Close closes the NotifySemaphore and all of its associated channels. It does not return until all semaphore channels have been closed. Calling Close on a closed NotifySemaphore returns an error.

func (*NotifySemaphore) Listen

func (s *NotifySemaphore) Listen(channel string) (<-chan *pq.Notification, error)

Listen starts listening on a notification channel. The returned Go channel ("semaphore channel") will be guaranteed to have at least one notification in it any time one or more notifications have been received from the database since the last receive on that channel.

It is not safe to call Listen if a concurrent Unlisten call on the same channel is in progress. However, it is safe to Listen on a channel which was previously Unlistened by a different goroutine.

If the channel is already active, pq.ErrChannelAlreadyOpen is returned. If the NotifySemaphore has been closed, an error is returned.

func (*NotifySemaphore) Ping

func (s *NotifySemaphore) Ping() error

Calls Ping() on the underlying Listener.

func (*NotifySemaphore) SetBroadcastOnPingTimeout

func (s *NotifySemaphore) SetBroadcastOnPingTimeout(broadcastOnPingTimeout bool)

Sets whether a nil *pq.Notification should be sent automatically when the server is pinged after inactivity.

func (*NotifySemaphore) SetPingInterval

func (s *NotifySemaphore) SetPingInterval(interval time.Duration)

Controls the amount of time the connection is allowed to stay idle before the server is pinged via Listener.Ping().

func (*NotifySemaphore) Unlisten

func (s *NotifySemaphore) Unlisten(channel string) error

Unlisten stops listening on the supplied notification channel and closes the semaphore channel associated with it. It is not safe to call Unlisten if a concurrent Listen call on that same channel is in progress, but it is safe to Unlisten a channel from a different goroutine than the one that previously executed Listen. It is also safe to call Unlisten while a goroutine is waiting on the semaphore channel. The channel will be closed gracefully.

Returns pq.ErrChannelNotOpen if the channel is not currently active, or an error if the NotifySemaphore has been closed.