MessageQueue.cc

Go to the documentation of this file.

/*
 * MessageQueue.cc
 *
 * Smalltalk like class library for C++
 * Message queue accessor.
 *
 * Copyright (c) 2003 Martin Dvorak, Milan Cermak
 */
/*
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
#include <stdio.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>

#include <stlib/os/MessageQueue.h>
#include <stlib/ByteArray.h>
#include <stlib/String.h>
#include "ipckey.h"

#define MAXBUFFERLEN 256

using namespace OS;

/* Attach message queue with given key */
MessageQueue::MessageQueue(int queueKey, bool exclusive)
{
    int flags = 0664 | IPC_CREAT;

    if (exclusive)
        flags = flags | IPC_EXCL;
    queueid = msgget(queueKey, flags);
    queueCreationKey = queueKey;
    if (queueid >= 0) active = true;
    _priority = 1;
}

MessageQueue *MessageQueue::open(void)
{
    return MessageQueue::open(new String("stlib-MessageQueue"));
}

MessageQueue *MessageQueue::open(bool exclusive)
{
    return MessageQueue::open(new String("stlib-MessageQueue"), exclusive);
}

MessageQueue *MessageQueue::open(String *key, bool exclusive)
{
    return new MessageQueue(getkey(key->asCString()), exclusive);
}

/* Accessing protocol */
long MessageQueue::bufferSize(void)
{
    return MAXBUFFERLEN;
}

void MessageQueue::priority(long level)
{
    _priority = level;
}

long MessageQueue::priority(void)
{
    return _priority;
}

/* Close active message queue */
void MessageQueue::close(void)
{
    if (active)
        msgctl(queueid, IPC_RMID, NULL);
    active = false;
}

/* Data transfer protocol */
int MessageQueue::readInto(ByteArray *buffer, long startIndex, long count)
{
    int bytes;
    struct msgbuf {
        long mtype;
        unsigned char data[MAXBUFFERLEN];
    } buf;

    if (queueid < 0) return 0;
    if (count > MAXBUFFERLEN) count = MAXBUFFERLEN;

    bytes = msgrcv(queueid, &buf, count, 0, MSG_NOERROR);
    if (bytes > 0)
        buffer->replace(startIndex, startIndex+bytes, buf.data);
    return bytes;
}

int MessageQueue::writeFrom(ByteArray *buffer, long startIndex, long count)
{
    struct msgbuf {
        long mtype;
        unsigned char mtext[MAXBUFFERLEN];
    } buf;

    if (queueid < 0 || count < 1)
        return 0;

    if (count > MAXBUFFERLEN) count = MAXBUFFERLEN;
    buf.mtype = _priority;
    /* Must memcpy, pointer doesn't work. */
    memcpy(buf.mtext, &(buffer->rawBytesReadOnly())[startIndex], count);
    int error = msgsnd(queueid, &buf, count, 0);
    if (error < 0)
        perror(__PRETTY_FUNCTION__);

    return (error < 0) ? 0 : count;
}

/* Positioning protocol */
void MessageQueue::seekTo(long position)
{
    shouldNotImplement(new String(__PRETTY_FUNCTION__));
}

---