Object
This class represents an POSIX message queue descriptor (mqd_t) object. It matches the C API for POSIX messages queues closely.
See the README for examples on how to use it.
An analogous Struct to “struct mq_attr” in C. This may be used in arguments for POSIX_MQ.new and POSIX_MQ#attr=. POSIX_MQ#attr returns an instance of this class.
See the mq_getattr(3) manpage for more information on the values.
The maximum number of open message descriptors supported by the system. This may be -1, in which case it is dynamically set at runtime. Consult your operating system documentation for system-specific information about this.
The maximum priority that may be specified for POSIX_MQ#send On POSIX-compliant systems, this is at least 31, but some systems allow higher limits. The minimum priority is always zero.
Opens a POSIX message queue given by name. name should start with a slash (“/”) for portable applications.
If a Symbol is given in place of integer flags, then:
:r is equivalent to IO::RDONLY
:w is equivalent to IO::CREAT|IO::WRONLY
:rw is equivalent to IO::CREAT|IO::RDWR
mode is an integer and only used when IO::CREAT is used. mq_attr is a POSIX_MQ::Attr and only used if IO::CREAT is used. If mq_attr is not specified when creating a queue, then the system defaults will be used.
See the manpage for mq_open(3) for more details on this function.
static VALUE init(int argc, VALUE *argv, VALUE self)
{
struct posix_mq *mq = get(self, 0);
struct open_args x;
VALUE name, oflags, mode, attr;
rb_scan_args(argc, argv, "13", &name, &oflags, &mode, &attr);
switch (TYPE(oflags)) {
case T_NIL:
x.oflags = O_RDONLY;
break;
case T_SYMBOL:
if (oflags == sym_r)
x.oflags = O_RDONLY;
else if (oflags == sym_w)
x.oflags = O_CREAT|O_WRONLY;
else if (oflags == sym_rw)
x.oflags = O_CREAT|O_RDWR;
else {
oflags = rb_inspect(oflags);
rb_raise(rb_eArgError,
"symbol must be :r, :w, or :rw: %s",
StringValuePtr(oflags));
}
break;
case T_BIGNUM:
case T_FIXNUM:
x.oflags = NUM2INT(oflags);
break;
default:
rb_raise(rb_eArgError, "flags must be an int, :r, :w, or :wr");
}
x.name = StringValueCStr(name);
x.argc = 2;
switch (TYPE(mode)) {
case T_FIXNUM:
x.argc = 3;
x.mode = NUM2UINT(mode);
break;
case T_NIL:
if (x.oflags & O_CREAT) {
x.argc = 3;
x.mode = 0666;
}
break;
default:
rb_raise(rb_eArgError, "mode not an integer");
}
switch (TYPE(attr)) {
case T_STRUCT:
x.argc = 4;
rstruct2mqattr(&x.attr, attr, 1);
/* principle of least surprise */
if (x.attr.mq_flags & O_NONBLOCK)
x.oflags |= O_NONBLOCK;
break;
case T_NIL:
break;
default:
check_struct_type(attr);
}
mq->des = (mqd_t)xopen(&x);
if (mq->des == MQD_INVALID) {
switch (errno) {
case ENOMEM:
case EMFILE:
case ENFILE:
case ENOSPC:
rb_gc();
mq->des = (mqd_t)xopen(&x);
}
if (mq->des == MQD_INVALID)
rb_sys_fail("mq_open");
}
mq->name = rb_str_dup(name);
if (x.oflags & O_NONBLOCK)
mq->attr.mq_flags = O_NONBLOCK;
return self;
}
Opens a POSIX message queue and performs operations on the given block, closing the message queue at exit. All all arguments are passed to POSIX_MQ.new.
Unlinks the message queue given by name. The queue will be destroyed when the last process with the queue open closes its queue descriptors.
static VALUE s_unlink(VALUE self, VALUE name)
{
int rv = mq_unlink(StringValueCStr(name));
if (rv == -1)
rb_sys_fail("mq_unlink");
return INT2NUM(1);
}
Inserts the given string into the message queue with a default priority of 0 and no timeout.
Returns itself so its calls may be chained. This use is only recommended only for users who expect blocking behavior from the queue.
static VALUE send0(VALUE self, VALUE buffer)
{
struct posix_mq *mq = get(self, 1);
struct rw_args x;
int rv;
setup_send_buffer(&x, buffer);
x.des = mq->des;
x.timeout = NULL;
x.msg_prio = 0;
retry:
rv = (int)rb_thread_blocking_region(xsend, &x, RUBY_UBF_IO, 0);
if (rv == -1) {
if (errno == EINTR)
goto retry;
rb_sys_fail("mq_send");
}
return self;
}
Returns a POSIX_MQ::Attr struct containing the attributes of the message queue. See the mq_getattr(3) manpage for more details.
static VALUE getattr(VALUE self)
{
struct posix_mq *mq = get(self, 1);
if (mq_getattr(mq->des, &mq->attr) < 0)
rb_sys_fail("mq_getattr");
return rb_funcall(cAttr, id_new, 4,
LONG2NUM(mq->attr.mq_flags),
LONG2NUM(mq->attr.mq_maxmsg),
LONG2NUM(mq->attr.mq_msgsize),
LONG2NUM(mq->attr.mq_curmsgs));
}
Only the IO::NONBLOCK flag may be set or unset (zero) in this manner. See the mq_setattr(3) manpage for more details.
Consider using the POSIX_MQ#nonblock= method as it is easier and more natural to use.
static VALUE setattr(VALUE self, VALUE astruct)
{
struct posix_mq *mq = get(self, 1);
struct mq_attr newattr;
rstruct2mqattr(&newattr, astruct, 0);
if (mq_setattr(mq->des, &newattr, NULL) < 0)
rb_sys_fail("mq_setattr");
return astruct;
}
There’s no point in ever cloning a POSIX_MQ object. All send/receive operations are atomic and only one native thread may be notified at a time
Closes the underlying message queue descriptor. If this descriptor had a registered notification request, the request will be removed so another descriptor or process may register a notification request. Message queue descriptors are automatically closed by garbage collection.
static VALUE _close(VALUE self)
{
struct posix_mq *mq = get(self, 1);
if (! MQ_IO_CLOSE(mq)) {
if (mq_close(mq->des) == -1)
rb_sys_fail("mq_close");
}
mq->des = MQD_INVALID;
return Qnil;
}
Returns true if the message queue descriptor is closed and therefore unusable, otherwise false
static VALUE closed(VALUE self)
{
struct posix_mq *mq = get(self, 0);
return mq->des == MQD_INVALID ? Qtrue : Qfalse;
}
There’s no point in ever duping a POSIX_MQ object. All send/receive operations are atomic and only one native thread may be notified at a time
Returns the string name of message queue associated with mq
static VALUE name(VALUE self)
{
struct posix_mq *mq = get(self, 0);
return rb_str_dup(mq->name);
}
Enables or disables non-blocking operation for the message queue descriptor. Errno::EAGAIN will be raised in situations where the queue would block. This is not compatible with timeout arguments to POSIX_MQ#send and POSIX_MQ#receive.
static VALUE setnonblock(VALUE self, VALUE nb)
{
struct mq_attr newattr;
struct posix_mq *mq = get(self, 1);
if (nb == Qtrue)
newattr.mq_flags = O_NONBLOCK;
else if (nb == Qfalse)
newattr.mq_flags = 0;
else
rb_raise(rb_eArgError, "must be true or false");
if (mq_setattr(mq->des, &newattr, &mq->attr) < 0)
rb_sys_fail("mq_setattr");
mq->attr.mq_flags = newattr.mq_flags;
return nb;
}
Returns the current non-blocking state of the message queue descriptor.
static VALUE nonblock_p(VALUE self)
{
struct posix_mq *mq = get(self, 1);
if (mq_getattr(mq->des, &mq->attr) < 0)
rb_sys_fail("mq_getattr");
return mq->attr.mq_flags & O_NONBLOCK ? Qtrue : Qfalse;
}
Executes the given block upon reception of the next message in an empty queue. If the message queue is not empty, then this block will only be fired after the queue is emptied and repopulated with one message.
This block will only be executed upon the arrival of the first message and must be reset/reenabled for subsequent notifications. This block will execute in a separate Ruby Thread (and thus will safely have the GVL by default).
This method is only supported on platforms that implement SIGEV_THREAD functionality in mq_notify(3). So far we only know of glibc + Linux supporting this. Please let us know if your platform can support this functionality and are willing to test for us <ruby.posix.mq@librelist.com>
As far as we can tell, this method is not very useful nor efficient. You would be better served using signals or just blocking. On Linux and FreeBSD, you can use POSIX_MQ with I/O multiplexing (IO.select, EventMachine), too.
Registers the notification request to deliver a given signal to the current process when message is received. If signal is nil, it will unregister and disable the notification request to allow other processes to register a request. If signal is false, it will register a no-op notification request which will prevent other processes from registering a notification. If signal is an IO object, it will spawn a thread upon the arrival of the next message and write one “\0” byte to the file descriptor belonging to that IO object. Only one process may have a notification request for a queue at a time, Errno::EBUSY will be raised if there is already a notification request registration for the queue.
Notifications are only fired once and processes must reregister for subsequent notifications.
For readers of the mq_notify(3) manpage, passing false is equivalent to SIGEV_NONE, and passing nil is equivalent of passing a NULL notification pointer to mq_notify(3).
static VALUE setnotify(VALUE self, VALUE arg)
{
struct posix_mq *mq = get(self, 1);
struct sigevent not;
struct sigevent * notification = ¬
VALUE rv = arg;
notify_cleanup(self);
not.sigev_notify = SIGEV_SIGNAL;
switch (TYPE(arg)) {
case T_FALSE:
not.sigev_notify = SIGEV_NONE;
break;
case T_NIL:
notification = NULL;
break;
case T_FIXNUM:
not.sigev_signo = NUM2INT(arg);
break;
case T_SYMBOL:
case T_STRING:
not.sigev_signo = lookup_sig(arg);
rv = INT2NUM(not.sigev_signo);
break;
default:
rb_raise(rb_eArgError, "must be a signal or nil");
}
my_mq_notify(mq->des, notification);
return rv;
}
Takes the highest priority message off the queue and returns an array containing the message as a String and the Integer priority of the message.
If the optional buffer is present, then it must be a String which will receive the data.
If the optional timeout is present, then it may be a Float or Integer specifying the timeout in seconds. Errno::ETIMEDOUT will be raised if timeout has elapsed and there are no messages in the queue.
On some older systems, the timeout argument is not currently supported and may raise NotImplementedError if timeout is used.
static VALUE receive(int argc, VALUE *argv, VALUE self)
{
return _receive(PMQ_WANTARRAY, argc, argv, self);
}
Inserts the given string into the message queue with an optional, unsigned integer priority. If the optional timeout is specified, then Errno::ETIMEDOUT will be raised if the operation cannot complete before timeout seconds has elapsed. Without timeout, this method may block until the queue is writable.
On some older systems, the timeout argument is not currently supported and may raise NotImplementedError if timeout is used.
static VALUE my_send(int argc, VALUE *argv, VALUE self)
{
return _send(0, argc, argv, self);
}
Takes the highest priority message off the queue and returns the message as a String.
If the optional buffer is present, then it must be a String which will receive the data.
If the optional timeout is present, then it may be a Float or Integer specifying the timeout in seconds. Errno::ETIMEDOUT will be raised if timeout has elapsed and there are no messages in the queue.
On some older systems, the timeout argument is not currently supported and may raise NotImplementedError if timeout is used.
static VALUE shift(int argc, VALUE *argv, VALUE self)
{
return _receive(0, argc, argv, self);
}
Returns an IO.select-able IO object. This method is only available under Linux and FreeBSD and is not intended to be portable.
static VALUE to_io(VALUE self)
{
struct posix_mq *mq = get(self, 1);
int fd = MQD_TO_FD(mq->des);
if (NIL_P(mq->io))
mq->io = rb_funcall(rb_cIO, id_new, 1, INT2NUM(fd));
return mq->io;
}
Exactly like POSIX_MQ#receive, except it returns nil instead of raising Errno::EAGAIN when non-blocking operation is desired.
This does not guarantee non-blocking behavior, the message queue must be made non-blocking before calling this method.
static VALUE tryreceive(int argc, VALUE *argv, VALUE self)
{
return _receive(PMQ_WANTARRAY|PMQ_TRY, argc, argv, self);
}
Exactly like POSIX_MQ#send, except it returns false instead of raising Errno::EAGAIN when non-blocking operation is desired and returns true on success instead of nil.
This does not guarantee non-blocking behavior, the message queue must be made non-blocking before calling this method.
static VALUE trysend(int argc, VALUE *argv, VALUE self)
{
return _send(PMQ_TRY, argc, argv, self);
}
Exactly like POSIX_MQ#shift, except it returns nil instead of raising Errno::EAGAIN when non-blocking operation is desired.
This does not guarantee non-blocking behavior, the message queue must be made non-blocking before calling this method.
static VALUE tryshift(int argc, VALUE *argv, VALUE self)
{
return _receive(PMQ_TRY, argc, argv, self);
}
Unlinks the message queue to prevent other processes from accessing it. All existing queue descriptors to this queue including those opened by other processes are unaffected. The queue will only be destroyed when the last process with open descriptors to this queue closes the descriptors.
static VALUE _unlink(VALUE self)
{
struct posix_mq *mq = get(self, 0);
int rv;
assert(TYPE(mq->name) == T_STRING && "mq->name is not a string");
rv = mq_unlink(RSTRING_PTR(mq->name));
if (rv == -1)
rb_sys_fail("mq_unlink");
return self;
}
Originally generated with the Darkfish Rdoc Generator 2, modified by wrongdoc.