go-ipc: bitbucket.org/avd/go-ipc/mq Index | Examples | Files

package mq

import "bitbucket.org/avd/go-ipc/mq"

Package mq implements interprocess queues logic. It provides access to system mq mechanisms, such as sysv mq and linux mq. Also, it provides access to multi-platform priority queue, FastMq.

Index

Examples

Package Files

doc.go mq.go mq_fast.go mq_fast_impl.go mq_helper_sysv.go mq_linux.go mq_sys_linux.go mq_sysv.go mq_sysv_sys.go mq_sysv_sys_linux+darwin.go shared_heap.go temp_error.go

Constants

const (
    // DefaultFastMqMaxSize is the default fast mq queue size.
    DefaultFastMqMaxSize = 8
    // DefaultFastMqMessageSize is the fast mq message size.
    DefaultFastMqMessageSize = 8192
)
const (
    // DefaultLinuxMqMaxSize is the default linux mq queue size.
    DefaultLinuxMqMaxSize = 8
    // DefaultLinuxMqMessageSize is the linux mq message size.
    // Its max value can be set via procfs.
    DefaultLinuxMqMessageSize = 8192
)
const (
    // O_NONBLOCK flag makes mq send/receive operations non-blocking.
    O_NONBLOCK = common.O_NONBLOCK
)

func Destroy Uses

func Destroy(name string) error

Destroy permanently removes mq object.

func DestroyFastMq Uses

func DestroyFastMq(name string) error

DestroyFastMq permanently removes a FastMq.

func DestroyLinuxMessageQueue Uses

func DestroyLinuxMessageQueue(name string) error

DestroyLinuxMessageQueue removes the queue permanently.

func DestroySystemVMessageQueue Uses

func DestroySystemVMessageQueue(name string) error

DestroySystemVMessageQueue permanently removes queue with a given name.

func FastMqAttrs Uses

func FastMqAttrs(name string) (int, int, error)

FastMqAttrs returns capacity and max message size of the existing mq.

func IsTemporary Uses

func IsTemporary(err error) bool

IsTemporary returns true, if an error is a timeout error.

func SetLinuxMqBlocking Uses

func SetLinuxMqBlocking(name string, block bool) error

SetLinuxMqBlocking sets whether the operations on a linux mq block. This will apply for all send/receive operations on any instance of the linux mq with the given name.

type Blocker Uses

type Blocker interface {
    SetBlocking(bool) error
}

Blocker is an object, which can work in blocking and non-blocking modes.

type Buffered Uses

type Buffered interface {
    Cap() int
}

Buffered is an object with internal buffer of the given capacity.

type FastMq Uses

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

FastMq is a priority message queue based on shared memory. Currently it is the only implementation for windows.

func CreateFastMq Uses

func CreateFastMq(name string, flag int, perm os.FileMode, maxQueueSize, maxMsgSize int) (*FastMq, error)

CreateFastMq creates new FastMq.

name - mq name. implementation will create a shm object with this name.
flag - flag is a combination of os.O_EXCL, and O_NONBLOCK.
perm - object's permission bits.
maxQueueSize - queue capacity.
maxMsgSize - maximum message size.

func OpenFastMq Uses

func OpenFastMq(name string, flag int) (*FastMq, error)

OpenFastMq opens an existing message queue. It returns an error, if it does not exist.

name - unique mq name.
flag - 0 or O_NONBLOCK.

func (*FastMq) Cap Uses

func (mq *FastMq) Cap() int

Cap returns size of the mq buffer.

func (*FastMq) Close Uses

func (mq *FastMq) Close() error

Close closes a FastMq instance.

func (*FastMq) Destroy Uses

func (mq *FastMq) Destroy() error

Destroy permanently removes a FastMq instance.

func (*FastMq) Empty Uses

func (mq *FastMq) Empty() bool

Empty returns true, if there are no messages in the queue.

func (*FastMq) Full Uses

func (mq *FastMq) Full() bool

Full returns true, if the capacity liimt has been reached.

func (*FastMq) Receive Uses

func (mq *FastMq) Receive(data []byte) (int, error)

Receive receives a message. It blocks if the queue is empty.

func (*FastMq) ReceivePriority Uses

func (mq *FastMq) ReceivePriority(data []byte) (int, int, error)

ReceivePriority receives a message and returns its priority. It blocks if the queue is empty.

func (*FastMq) ReceivePriorityTimeout Uses

func (mq *FastMq) ReceivePriorityTimeout(data []byte, timeout time.Duration) (int, int, error)

ReceivePriorityTimeout receives a message and returns its priority. It blocks if the queue is empty, waiting for not longer, then the timeout.

func (*FastMq) ReceiveTimeout Uses

func (mq *FastMq) ReceiveTimeout(data []byte, timeout time.Duration) (int, error)

ReceiveTimeout receives a message. It blocks if the queue is empty. It blocks if the queue is empty, waiting for not longer, then the timeout.

func (*FastMq) Send Uses

func (mq *FastMq) Send(data []byte) error

Send sends a message. It blocks if the queue is full.

func (*FastMq) SendPriority Uses

func (mq *FastMq) SendPriority(data []byte, prio int) error

SendPriority sends a message with the given priority. It blocks if the queue is full.

func (*FastMq) SendPriorityTimeout Uses

func (mq *FastMq) SendPriorityTimeout(data []byte, prio int, timeout time.Duration) error

SendPriorityTimeout sends a message with the given priority. It blocks if the queue is full, waiting for not longer, then the timeout.

func (*FastMq) SendTimeout Uses

func (mq *FastMq) SendTimeout(data []byte, timeout time.Duration) error

SendTimeout sends a message with the default priority 0. It blocks if the queue is full, waiting for not longer, then the timeout.

func (*FastMq) SetBlocking Uses

func (mq *FastMq) SetBlocking(block bool) error

SetBlocking sets whether the send/receive operations on the queue block. This applies to the current instance only.

type LinuxMessageQueue Uses

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

LinuxMessageQueue is a linux-specific ipc mechanism based on message passing.

func CreateLinuxMessageQueue Uses

func CreateLinuxMessageQueue(name string, flag int, perm os.FileMode, maxQueueSize, maxMsgSize int) (*LinuxMessageQueue, error)

CreateLinuxMessageQueue creates new queue with the given name and permissions.

name - unique mq name.
flag - flag is a combination of os.O_EXCL and O_NONBLOCK.
perm - object's permission bits.
maxQueueSize - queue capacity.
maxMsgSize - maximum message size.

func OpenLinuxMessageQueue Uses

func OpenLinuxMessageQueue(name string, flag int) (*LinuxMessageQueue, error)

OpenLinuxMessageQueue opens an existing message queue. It returns an error, if it does not exist.

name - unique mq name.
flag - flag is a combination of (os.O_RDONLY or os.O_WRONLY or os.O_RDWR) and O_NONBLOCK.
	O_RDONLY
		Open the queue to receive messages only.
	O_WRONLY
		Open the queue to send messages only.
	O_RDWR
		Open the queue to both send and receive messages.

func (*LinuxMessageQueue) Cap Uses

func (mq *LinuxMessageQueue) Cap() int

Cap returns the size of the mq buffer.

func (*LinuxMessageQueue) Close Uses

func (mq *LinuxMessageQueue) Close() error

Close closes the queue.

func (*LinuxMessageQueue) Destroy Uses

func (mq *LinuxMessageQueue) Destroy() error

Destroy closes the queue and removes it permanently.

func (*LinuxMessageQueue) ID Uses

func (mq *LinuxMessageQueue) ID() int

ID returns unique id of the queue.

func (*LinuxMessageQueue) Notify Uses

func (mq *LinuxMessageQueue) Notify(ch chan<- int) error

Notify notifies about new messages in the queue by sending id of the queue to the channel. If there are messages in the queue, no notification will be sent unless all of them are read.

func (*LinuxMessageQueue) NotifyCancel Uses

func (mq *LinuxMessageQueue) NotifyCancel() error

NotifyCancel cancels notification subscription.

func (*LinuxMessageQueue) Receive Uses

func (mq *LinuxMessageQueue) Receive(data []byte) (int, error)

Receive receives a message. It blocks if the queue is empty. Returns message len.

func (*LinuxMessageQueue) ReceivePriority Uses

func (mq *LinuxMessageQueue) ReceivePriority(data []byte) (int, int, error)

ReceivePriority receives a message, returning its priority. It blocks if the queue is empty. Returns message len and priority.

func (*LinuxMessageQueue) ReceiveTimeout Uses

func (mq *LinuxMessageQueue) ReceiveTimeout(data []byte, timeout time.Duration) (int, error)

ReceiveTimeout receives a message. It blocks if the queue is empty, waiting for a message unless timeout is passed. Returns message len.

func (*LinuxMessageQueue) ReceiveTimeoutPriority Uses

func (mq *LinuxMessageQueue) ReceiveTimeoutPriority(input []byte, timeout time.Duration) (int, int, error)

ReceiveTimeoutPriority receives a message, returning its priority. It blocks if the queue is empty, waiting for a message unless timeout is passed. Returns message len and priority.

func (*LinuxMessageQueue) Send Uses

func (mq *LinuxMessageQueue) Send(data []byte) error

Send sends a message with a default (0) priority. It blocks if the queue is full.

func (*LinuxMessageQueue) SendPriority Uses

func (mq *LinuxMessageQueue) SendPriority(data []byte, prio int) error

SendPriority sends a message with a given priority. It blocks if the queue is full.

func (*LinuxMessageQueue) SendTimeout Uses

func (mq *LinuxMessageQueue) SendTimeout(data []byte, timeout time.Duration) error

SendTimeout sends a message with a default (0) priority. It blocks if the queue is full, waiting for a message unless timeout is passed.

func (*LinuxMessageQueue) SendTimeoutPriority Uses

func (mq *LinuxMessageQueue) SendTimeoutPriority(data []byte, prio int, timeout time.Duration) error

SendTimeoutPriority sends a message with a given priority. It blocks if the queue is full, waiting for a message unless timeout is passed.

func (*LinuxMessageQueue) SetBlocking Uses

func (mq *LinuxMessageQueue) SetBlocking(block bool) error

SetBlocking sets whether the send/receive operations on the queue block. This applies to the current instance only.

type Messenger Uses

type Messenger interface {
    // Send sends the data. It blocks if there are no readers and the queue is full
    Send(data []byte) error
    // Receive reads data from the queue. It blocks if the queue is empty.
    // Returns message len.
    Receive(data []byte) (int, error)
    io.Closer
}

Messenger is an interface which must be satisfied by any message queue implementation on any platform.

Code:

Destroy("mq")
mq, err := New("mq", os.O_CREATE|os.O_EXCL, 0666)
if err != nil {
    panic("new queue")
}
defer mq.Close()
data := []byte{1, 2, 3, 4, 5, 6, 7, 8}
go func() {
    if err := mq.Send(data); err != nil {
        panic("send")
    }
}()
mq2, err := Open("mq", 0)
if err != nil {
    panic("open")
}
defer mq2.Close()
received := make([]byte, len(data))
l, err := mq2.Receive(received)
if err != nil {
    panic("receive")
}
if l != len(data) {
    panic("wrong len")
}
for i, b := range received {
    if b != data[i] {
        panic("wrong data")
    }
}

func New Uses

func New(name string, flag int, perm os.FileMode) (Messenger, error)

New creates a mq with a given name and permissions. It uses the default implementation. If there are several implementations on a platform, you can use explicit create functions.

name - unique queue name.
flag - create flags. You can specify:
	os.O_EXCL if you don't want to open a queue if it exists.
	O_NONBLOCK if you don't want to block on send/receive.
		This flag may not be supported by a particular implementation. To be sure, you can convert Messenger
		to Blocker and call SetBlocking to set/unset non-blocking mode.
perm - permissions for the new queue.

func Open Uses

func Open(name string, flags int) (Messenger, error)

Open opens a mq with a given name and flags. It uses the default implementation. If there are several implementations on a platform, you can use explicit create functions.

name  - unique queue name.
flags - 0 or O_NONBLOCK.

type PriorityMessenger Uses

type PriorityMessenger interface {
    Messenger
    Buffered
    // SendPriority sends the data. The message will be inserted in the mq according to its priority.
    SendPriority(data []byte, prio int) error
    // ReceivePriority reads a message and returns its len and priority.
    ReceivePriority(data []byte) (int, int, error)
}

PriorityMessenger is a Messenger, which orders messages according to their priority. Semantic is similar to linux native mq: Messages are placed on the queue in decreasing order of priority, with newer messages of the same priority being placed after older messages with the same priority.

Code:

Destroy("mq")
mq, err := New("mq", os.O_CREATE|os.O_EXCL, 0666)
if err != nil {
    panic("new queue")
}
defer mq.Close()
// not all implementations support prioritized send/receive.
tmq, ok := mq.(PriorityMessenger)
if !ok {
    panic("not a prio messenger")
}
data := []byte{1, 2, 3, 4, 5, 6, 7, 8}
go func() {
    if err := tmq.SendPriority(data, 0); err != nil {
        panic("send")
    }
    if err := tmq.SendPriority(data, 1); err != nil {
        panic("send")
    }
}()
mq2, err := Open("mq", 0)
if err != nil {
    panic("open")
}
defer mq2.Close()
tmq2, ok := mq2.(PriorityMessenger)
if !ok {
    panic("not a prio messenger")
}
received := make([]byte, len(data))
_, prio, err := tmq2.ReceivePriority(received)
if err != nil || prio != 1 {
    panic("receive")
}
_, prio, err = tmq2.ReceivePriority(received)
if err != nil || prio != 0 {
    panic("receive")
}

type SystemVMessageQueue Uses

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

SystemVMessageQueue is a System V ipc mechanism based on message passing.

func CreateSystemVMessageQueue Uses

func CreateSystemVMessageQueue(name string, flag int, perm os.FileMode) (*SystemVMessageQueue, error)

CreateSystemVMessageQueue creates new queue with the given name and permissions.

name - unique mq name.
flag - flag is a combination of os.O_EXCL and O_NONBLOCK.
perm - object's permission bits.

func OpenSystemVMessageQueue Uses

func OpenSystemVMessageQueue(name string, flags int) (*SystemVMessageQueue, error)

OpenSystemVMessageQueue opens existing message queue.

name - unique mq name.
flag - 0 and O_NONBLOCK.

func (*SystemVMessageQueue) Close Uses

func (mq *SystemVMessageQueue) Close() error

Close closes the queue. As there is no need to close SystemV mq, this function returns nil. It was added to satisfy io.Closer

func (*SystemVMessageQueue) Destroy Uses

func (mq *SystemVMessageQueue) Destroy() error

Destroy closes the queue and removes it permanently.

func (*SystemVMessageQueue) Receive Uses

func (mq *SystemVMessageQueue) Receive(data []byte) (int, error)

Receive receives a message. It blocks if the queue is empty.

func (*SystemVMessageQueue) Send Uses

func (mq *SystemVMessageQueue) Send(data []byte) error

Send sends a message. It blocks if the queue is full.

func (*SystemVMessageQueue) SetBlocking Uses

func (mq *SystemVMessageQueue) SetBlocking(block bool) error

SetBlocking sets whether the send/receive operations on the queue block.

type TimedMessenger Uses

type TimedMessenger interface {
    Messenger
    // SendTimeout sends the data. It blocks if there are no readers and the queue if full.
    // It waits for not more, than timeout.
    SendTimeout(data []byte, timeout time.Duration) error
    // ReceiveTimeout reads data from the queue. It blocks if the queue is empty.
    // It waits for not more, than timeout. Returns message len.
    ReceiveTimeout(data []byte, timeout time.Duration) (int, error)
}

TimedMessenger is a Messenger, which supports send/receive timeouts. Passing 0 as a timeout makes a call non-blocking. Passing negative value as a timeout makes the timeout infinite.

Code:

Destroy("mq")
mq, err := New("mq", os.O_CREATE|os.O_EXCL, 0666)
if err != nil {
    panic("new queue")
}
defer mq.Close()
data := []byte{1, 2, 3, 4, 5, 6, 7, 8}
go func() {
    // send after [0..500] ms delay.
    time.Sleep(time.Duration((rand.Int() % 6)) * time.Millisecond * 100)
    if err := mq.Send(data); err != nil {
        panic("send")
    }
}()
mq2, err := Open("mq", 0)
if err != nil {
    panic("open")
}
defer mq2.Close()
// not all implementations support timed send/receive.
tmq, ok := mq2.(TimedMessenger)
if !ok {
    panic("not a timed messenger")
}
received := make([]byte, len(data))
// depending on send delay we either get a timeout error, or receive the data.
l, err := tmq.ReceiveTimeout(received, 500*time.Millisecond)
if err != nil {
    if !IsTemporary(err) {
        panic(err)
    } else { // handle timeout.
        return
    }
}
if l != len(data) {
    panic("wrong len")
}
for i, b := range received {
    if b != data[i] {
        panic("wrong data")
    }
}

Package mq imports 18 packages (graph). Updated 2019-05-07. Refresh now. Tools for package owners.