/*
* cmslib_client.c: SAForum AIS Message Service client library
*
* Copyright (c) 2004 Intel Corp.
*
* Author: Zou Yixiong (yixiong.zou@intel.com)
* Author: Zhu Yi (yi.zhu@intel.com)
*
* 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 <stdlib.h>
#include <unistd.h> /* dup, dup2 */
#include <string.h>
#include <assert.h>
#include <clplumbing/realtime.h>
#include <cl_log.h>
#include <heartbeat.h>
#include <saf/ais.h>
#include "cmsclient.h"
#include "cmsclient_types.h"
#define PIPETRICK_DEBUG 0
#ifdef DEBUG_LIBRARY
#define dprintf(arg...) fprintf(stderr, ##arg)
#else
#define dprintf(arg...) {}
#endif
#define GET_CMS_HANDLE(x) ((x == NULL) ? NULL : \
(__cms_handle_t *)g_hash_table_lookup( \
__cmshandle_hash, x))
#define GET_MQ_HANDLE(x) ((x == NULL) ? NULL : \
(__cms_handle_t *)g_hash_table_lookup( \
__mqhandle_hash, x))
static GHashTable * __cmshandle_hash;
static GHashTable * __mqhandle_hash;
static GHashTable * __group_tracking_hash;
static guint __cmshandle_counter = 0;
static gboolean __cmsclient_init_flag = FALSE;
static gboolean __notify_acked = TRUE;
void cmsclient_hash_init(void);
IPC_Channel *cms_channel_conn(void);
int enqueue_dispatch_msg(__cms_handle_t * hd, client_header_t * msg);
client_header_t * dequeue_dispatch_msg(GList ** queue);
int read_and_queue_ipc_msg(__cms_handle_t * handle);
int dispatch_msg(__cms_handle_t * handle, client_header_t * msg);
int wait_for_msg(__cms_handle_t * handle, size_t msgtype,
const SaNameT * name, client_header_t ** msg, SaTimeT timeout);
int get_timeout_value(SaTimeT timeout, struct timeval * tv);
static int
saname_cmp(const SaNameT s1, const SaNameT s2)
{
SaUint16T len1, len2;
/* dprintf("Length of s1: %d, s2: %d\n", s1.length, s2.length); */
len1 = s1.value[s1.length - 1] ? s1.length : s1.length - 1;
len2 = s2.value[s2.length - 1] ? s2.length : s2.length - 1;
if (len1 != len2)
return len2 - len1;
return strncmp(s1.value, s2.value, len1);
}
static int
bad_saname(const SaNameT * name)
{
int i;
if (!name || name->length <= 0
|| name->length > SA_MAX_NAME_LENGTH - 1)
return TRUE;
/*
* We don't support '\0' inside a SaNameT.value.
*/
for (i = 0; i < name->length; i++)
if (name->value[i] == '\0')
return TRUE;
return FALSE;
}
static char *
saname2str(SaNameT name)
{
char * str;
if (name.length <= 0)
return NULL;
if (name.length > SA_MAX_NAME_LENGTH - 1)
name.length = SA_MAX_NAME_LENGTH - 1;
if ((str = (char *)ha_malloc(name.length + 1)) == NULL)
return NULL;
strncpy(str, name.value, name.length);
str[name.length] = '\0';
return str;
}
static int
active_poll(__cms_handle_t * hd)
{
int fd;
if (hd->backup_fd >= 0) {
cl_log(LOG_WARNING, "%s: recursion detected", __FUNCTION__);
return 1;
}
if ((fd = hd->ch->ops->get_recv_select_fd(hd->ch)) < 0) {
cl_log(LOG_ERR, "%s: get_recv_select_fd failed", __FUNCTION__);
return 1;
}
if ((hd->backup_fd = dup(fd)) == -1) {
cl_log(LOG_ERR, "%s: dup2 failed", __FUNCTION__);
perror("dup2");
return 1;
}
close(fd);
if (dup2(hd->active_fd, fd) == -1) {
cl_log(LOG_ERR, "%s: dup2 failed", __FUNCTION__);
perror("dup2");
return 1;
}
#if PIPETRICK_DEBUG
dprintf("acitve_poll for <%p>\n", hd);
#endif
return 0;
}
static int
restore_poll(__cms_handle_t * hd)
{
int fd;
if (hd->backup_fd < 0) {
cl_log(LOG_WARNING, "%s: recursion detected", __FUNCTION__);
return 1;
}
if ((fd = hd->ch->ops->get_recv_select_fd(hd->ch)) < 0) {
cl_log(LOG_ERR, "%s: get_recv_select_fd failed", __FUNCTION__);
return 1;
}
if (dup2(hd->backup_fd, fd) == -1) {
cl_log(LOG_ERR, "%s: dup2 failed", __FUNCTION__);
return 1;
}
hd->backup_fd = -1; /* mark as unused */
#if PIPETRICK_DEBUG
dprintf("restore_poll for <%p>\n", hd);
#endif
return 0;
}
static int
cmsclient_message_recv(__cms_handle_t * hd, client_header_t ** data)
{
int ret;
IPC_Message * ipc_msg;
void * buf;
if (hd->backup_fd >= 0)
restore_poll(hd);
ret = hd->ch->ops->recv(hd->ch, &ipc_msg);
if (ret != IPC_OK)
return ret;
buf = ha_malloc(ipc_msg->msg_len);
if (!buf) {
cl_log(LOG_CRIT, "unable to malloc data for cmsclient_message_recv.");
return IPC_FAIL;
}
memcpy(buf, ipc_msg->msg_body, ipc_msg->msg_len);
*data = (client_header_t *) buf;
ipc_msg->msg_done(ipc_msg);
return ret;
}
static void
cmsclient_message_done(IPC_Message * msg)
{
char * name;
client_header_t * message;
message = msg->msg_body;
name = saname2str(message->name);
ha_free(msg->msg_private);
ha_free(name);
}
static int
cmsclient_message_send(__cms_handle_t * hd, size_t len, gpointer data)
{
IPC_Message * msg;
if ((msg = ha_malloc(sizeof(IPC_Message) + len)) == NULL) {
cl_log(LOG_ERR, "%s: ha_malloc failed", __FUNCTION__);
return FALSE;
}
if (hd->backup_fd >= 0)
restore_poll(hd);
msg->msg_body = msg + 1;
memcpy(msg->msg_body, data, len);
msg->msg_len = len;
msg->msg_private = msg;
msg->msg_done = cmsclient_message_done;
return hd->ch->ops->send(hd->ch, msg);
}
static gboolean
msgqueue_remove(gpointer key, gpointer value, gpointer user_data)
{
__cms_queue_handle_t * qhd = (__cms_queue_handle_t *) value;
client_mqueue_close_t cmg;
SaNameT * qname;
CMS_LIBRARY_TRACE();
qname = &(qhd->queue_name);
cmg.header.type = CMS_QUEUE_CLOSE;
cmg.header.name = *qname;
cmg.handle = qhd->queue_handle;
cmg.silent = TRUE;
cmsclient_message_send(qhd->cms_handle, sizeof(cmg), &cmg);
g_hash_table_remove(__mqhandle_hash, key);
ha_free((__cms_queue_handle_t *) qhd);
return TRUE;
}
static gboolean
library_initialized(void)
{
return __cmsclient_init_flag;
}
void
cmsclient_hash_init()
{
if (library_initialized())
return;
__cmshandle_hash = g_hash_table_new(g_int_hash, g_int_equal);
__mqhandle_hash = g_hash_table_new(g_int_hash, g_int_equal);
__group_tracking_hash = g_hash_table_new(g_str_hash, g_str_equal);
__cmsclient_init_flag = TRUE;
}
/*
* This is a blocking wait for a particular type of msg on a particular queue.
* Note: memory allocated in this function. caller needs to free().
*/
int
wait_for_msg(__cms_handle_t * handle, size_t msgtype,
const SaNameT * queueName, client_header_t ** msg,
SaTimeT timeout)
{
int fd;
client_header_t * cms_msg;
longclock_t t_start = 0, t_end = 0;
if (timeout < 0)
return SA_ERR_INVALID_PARAM;
if (timeout != SA_TIME_END) {
t_start = time_longclock();
t_end = t_start + msto_longclock(timeout/1000);
}
if (handle->backup_fd >= 0)
restore_poll(handle);
fd = handle->ch->ops->get_recv_select_fd(handle->ch);
dprintf("In %s for message type 0x%x\n", __FUNCTION__, msgtype);
while (1) {
int ret = -1;
struct timeval * tv, to;
fd_set rset;
FD_ZERO(&rset);
FD_SET(fd, &rset);
tv = NULL;
if (timeout != SA_TIME_END) {
to.tv_sec = longclockto_ms((t_end - t_start))/1000;
to.tv_usec = (((t_end - t_start) -
secsto_longclock(to.tv_sec)))/1000;
tv = &to;
}
if (!handle->ch->ops->is_message_pending(handle->ch)
&& (ret = select(fd + 1, &rset, NULL, NULL, tv)) == -1) {
cl_log(LOG_ERR, "%s: select error", __FUNCTION__);
return SA_ERR_LIBRARY;
} else if (ret == 0) {
cl_log(LOG_WARNING, "%s: timeout!", __FUNCTION__);
return SA_ERR_TIMEOUT;
}
if ((ret = cmsclient_message_recv(handle, &cms_msg))!= IPC_OK) {
if (ret == IPC_FAIL) {
cl_shortsleep();
continue;
}
cl_log(LOG_ERR, "%s: cmsclient_message_recv failed, "
"rc = %d", __FUNCTION__, ret);
return SA_ERR_LIBRARY;
}
dprintf("received msg... type = 0x%x, name = %s\n"
, cms_msg->type, cms_msg->name.value);
if (cms_msg->type & msgtype && queueName != NULL
&& queueName != NULL
&& (saname_cmp(cms_msg->name, *queueName) == 0)) {
dprintf("Func %s return OK\n", __FUNCTION__);
*msg = cms_msg;
if (g_list_length(handle->dispatch_queue))
active_poll(handle);
return SA_OK;
} else {
enqueue_dispatch_msg(handle, cms_msg);
t_start = time_longclock();
}
}
}
IPC_Channel *
cms_channel_conn(void)
{
IPC_Channel * ch;
GHashTable * attrs;
char path[] = IPC_PATH_ATTR;
char cms_socket[] = CMS_DOMAIN_SOCKET;
int ret;
attrs = g_hash_table_new(g_str_hash,g_str_equal);
g_hash_table_insert(attrs, path, cms_socket);
ch = ipc_channel_constructor(IPC_DOMAIN_SOCKET, attrs);
g_hash_table_destroy(attrs);
if (ch ) {
ret = ch->ops->initiate_connection(ch);
if (ret != IPC_OK) {
cl_log(LOG_ERR, "cms_channel_conn failed, maybe "
"you don't have cms server running...");
return NULL;
}
return ch;
}
else
return NULL;
}
static int
enqueue_dispatch_item(GList **queue, client_header_t * item)
{
*queue = g_list_append(*queue, item);
return SA_OK;
}
int
enqueue_dispatch_msg(__cms_handle_t * hd, client_header_t * msg)
{
client_message_t * fmsg = (client_message_t *)msg;
dprintf("calling enqueue_dispatch_msg ..... \n");
/*
* If it is a message, then add it to the msg queue.
*/
if (msg->type == CMS_MSG_NOTIFY) {
dprintf("got a CMS_MSG_NOTIFY msg\n");
__notify_acked = FALSE;
}
return enqueue_dispatch_item(&(hd->dispatch_queue),
(client_header_t *) fmsg);
}
client_header_t *
dequeue_dispatch_msg(GList ** queue)
{
client_header_t * msg = NULL;
GList * head;
if (!g_list_length(*queue))
return NULL;
head = g_list_first(*queue);
*queue = g_list_remove_link(*queue, head);
msg = head->data;
g_list_free_1(head);
return msg;
}
/**
* Read all the ipc msg in the buffer and queue them to
* the msg queue or the dispatch queue.
*/
int
read_and_queue_ipc_msg(__cms_handle_t * handle)
{
int ret, count = 0;
client_header_t *rcmg;
__mqgroup_track_t * track;
client_mqgroup_notify_t *nsg, *m;
dprintf("b4 the do loop of the read_and_queue_ipc_msg ...\n");
if (handle->backup_fd >= 0)
restore_poll(handle);
while (handle->ch->ops->is_message_pending(handle->ch)) {
ret = cmsclient_message_recv(handle, &rcmg);
if (ret == IPC_FAIL) {
cl_shortsleep();
cl_log(LOG_WARNING, "%s: cmsclient_message_recv "
"failed, rc = %d", __FUNCTION__, ret);
break;
}
switch (rcmg->type) {
case CMS_QUEUEGROUP_NOTIFY:
/*
* prepare the notify buffer
*/
m = (client_mqgroup_notify_t *)rcmg;
m->data = (char *)rcmg +
sizeof(client_mqgroup_notify_t);
track = g_hash_table_lookup(__group_tracking_hash,
(m->group_name).value);
if (track == NULL) {
/*
* This is possible, because TrackStop
* may be called before we get here.
*/
cl_log(LOG_INFO, "No one tracks the group"
" [%s] membership now!"
, m->group_name.value);
return TRUE;
}
track->policy = m->policy;
/*
* only enqueue head is enough for us
*/
dprintf("enqueue group notify msg head\n");
nsg = (client_mqgroup_notify_t *)
ha_malloc(sizeof(client_mqgroup_notify_t));
if (!nsg) {
ha_free(rcmg);
cl_log(LOG_CRIT, "malloc failed for the notification.");
return FALSE;
}
memcpy(nsg, m, sizeof(client_mqgroup_notify_t));
enqueue_dispatch_msg(handle, (client_header_t *)nsg);
ha_free(rcmg);
break;
default:
enqueue_dispatch_msg(handle, rcmg);
/* TODO: we have a memory leak here
need to call the msg_done()
*/
break;
}
}
return count;
}
int
dispatch_msg(__cms_handle_t * handle, client_header_t * msg)
{
client_mqueue_open_t * omsg;
client_mqgroup_notify_t * nmsg;
client_message_ack_t * amsg;
__mqgroup_track_t * track;
client_message_t * gmsg;
char * name;
__cms_queue_handle_t * qhd;
dprintf("In Function %s..\n", __FUNCTION__);
dprintf("handle=<%p> msg->type=<%d>\n", handle, msg->type);
if (handle == NULL || msg == NULL)
return HA_FAIL;
switch (msg->type) {
case CMS_QUEUE_OPEN_ASYNC:
omsg = (client_mqueue_open_t *) msg;
if ((handle->callbacks).saMsgQueueOpenCallback) {
if (omsg->header.flag != SA_OK) {
omsg->handle = 0;
}
(handle->callbacks).saMsgQueueOpenCallback(
&(omsg->handle), omsg->invocation,
omsg->header.flag);
}
ha_free(omsg);
break;
case CMS_MSG_NOTIFY:
gmsg = (client_message_t *) msg;
qhd = g_hash_table_lookup(handle->queue_handle_hash,
&(gmsg->handle));
if (handle->callbacks.saMsgMessageReceivedCallback)
handle->callbacks.saMsgMessageReceivedCallback(
&(qhd->queue_handle), NULL, 0);
ha_free(gmsg);
break;
case CMS_MSG_ACK:
amsg = (client_message_ack_t *) msg;
if ((handle->callbacks).saMsgMessageDeliveredCallback) {
(handle->callbacks).saMsgMessageDeliveredCallback(
amsg->invocation,
msg->flag);
}
ha_free(amsg);
break;
case CMS_QUEUEGROUP_NOTIFY:
nmsg = (client_mqgroup_notify_t *)msg;
name = (char *) ha_malloc(nmsg->group_name.length + 1);
if (name == NULL) {
cl_log(LOG_ERR, "%s: ha_malloc failed", __FUNCTION__);
return FALSE;
}
dprintf("group name [%s], length [%d]\n"
, nmsg->group_name.value, nmsg->group_name.length);
strncpy(name, nmsg->group_name.value, nmsg->group_name.length);
name[nmsg->group_name.length] = '\0';
dprintf("name = [%s]\n", name);
track = g_hash_table_lookup(__group_tracking_hash, name);
if (track == NULL) {
cl_log(LOG_ERR, "Cannot find track buffer");
return FALSE;
}
if ((handle->callbacks).saMsgQueueGroupTrackCallback == NULL)
return FALSE;
ha_free(name);
ha_free(nmsg);
break;
default:
return HA_FAIL;
}
return HA_OK;
}
SaErrorT
saMsgInitialize(SaMsgHandleT *msgHandle, const SaMsgCallbacksT *msgCallbacks,
const SaVersionT *version)
{
IPC_Channel *ch;
__cms_handle_t *hd;
SaMsgHandleT * key;
int pipefd[2];
cl_log_set_entity("libcms");
cl_log_set_facility(LOG_USER);
#ifdef DEBUG_LIBRARY
cl_log_enable_stderr(TRUE);
#endif
if ((!version)
|| version->releaseCode < 'A' || version->releaseCode > 'Z'
|| (version->releaseCode << 8) + (version->major << 4) +
version->minor > (AIS_VERSION_RELEASE_CODE << 8) +
(AIS_VERSION_MAJOR << 4) + AIS_VERSION_MINOR) {
cl_log(LOG_ERR, "AIS library version is lower then required");
return SA_ERR_VERSION;
}
if (!msgHandle)
return SA_ERR_INVALID_PARAM;
if (!(ch = cms_channel_conn())) {
cl_log(LOG_ERR, "cms_channel_conn failed.");
return SA_ERR_LIBRARY;
}
if (pipe(pipefd) == -1) {
cl_log(LOG_ERR, "create pipe failed");
return SA_ERR_LIBRARY;
}
/*
* Write something to the pipe but we never read so that
* select to this fd will always return immediately.
*/
if (write(pipefd[1], "ACTIVE", 6) < 0) {
cl_log(LOG_ERR, "write pipe failed\n");
return SA_ERR_LIBRARY;
}
cmsclient_hash_init();
dprintf("ch_status = %d\n", ch->ch_status);
dprintf("farside_pid = %d\n", ch->farside_pid);
hd = (__cms_handle_t *)ha_malloc(sizeof(__cms_handle_t));
if (!hd)
return SA_ERR_NO_MEMORY;
memset(hd, 0, sizeof(__cms_handle_t));
hd->queue_handle_hash = g_hash_table_new(g_int_hash, g_int_equal);
hd->ch = ch;
if (msgCallbacks) {
memcpy(&(hd->callbacks), msgCallbacks, sizeof(SaMsgCallbacksT));
} else {
memset(&(hd->callbacks), 0, sizeof(SaMsgCallbacksT));
}
*msgHandle = __cmshandle_counter++;
hd->service_handle = *msgHandle;
hd->active_fd = pipefd[0];
hd->backup_fd = -1;
key = (SaMsgHandleT *) ha_malloc(sizeof(SaMsgHandleT));
key = msgHandle;
g_hash_table_insert(__cmshandle_hash, key, hd);
return SA_OK;
}
SaErrorT
saMsgFinalize(SaMsgHandleT *msgHandle)
{
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
g_hash_table_foreach_remove(hd->queue_handle_hash, msgqueue_remove, hd);
g_hash_table_remove(__cmshandle_hash, msgHandle);
if (hd->backup_fd >= 0)
restore_poll(hd);
hd->ch->ops->destroy(hd->ch);
close(hd->active_fd);
/* TODO: need to free the glist on the dispatch queue */
ha_free(hd);
return SA_OK;
}
SaErrorT
saMsgQueueOpen(const SaMsgHandleT *msgHandle,
const SaNameT *queueName,
const SaMsgQueueCreationAttributesT *creationAttributes,
SaMsgQueueOpenFlagsT openFlags,
SaMsgQueueHandleT *queueHandle,
SaTimeT timeout)
{
int ret;
client_mqueue_open_t cmg;
client_mqueue_open_t *rcmg;
client_header_t * reply;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (!openFlags)
return SA_ERR_BAD_FLAGS;
if (bad_saname(queueName) || !queueHandle
|| (!creationAttributes && !(openFlags & SA_MSG_QUEUE_OPEN_ONLY))
|| (creationAttributes && (openFlags & SA_MSG_QUEUE_OPEN_ONLY)))
return SA_ERR_INVALID_PARAM;
cmg.header.type = CMS_QUEUE_OPEN;
cmg.header.name = *queueName;
if (creationAttributes) {
if ((creationAttributes->creationFlags != 0)
&& creationAttributes->creationFlags !=
SA_MSG_QUEUE_PERSISTENT)
return SA_ERR_BAD_FLAGS;
cmg.attr = *creationAttributes;
} else {
/*
* else set to -1, so that daemon knows client didn't
* provide a creationAttributes
*/
cmg.attr.creationFlags = -1;
}
cmg.openflag = openFlags;
cmg.invocation = 0;
cmg.policy = 0;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
if (openFlags & SA_MSG_QUEUE_RECEIVE_CALLBACK
&& !(hd->callbacks).saMsgQueueOpenCallback
&& !(hd->callbacks).saMsgMessageReceivedCallback)
return SA_ERR_INIT;
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
dprintf("%s: cmsclient_message_send returns %d\n", __FUNCTION__, ret);
/*
* We should only have one client blocking for it.
*/
ret = wait_for_msg(hd, CMS_QUEUE_OPEN, queueName, &reply, timeout);
if (ret != SA_OK) {
return ret;
}
rcmg = (client_mqueue_open_t *) reply;
if ((ret = (rcmg->header).flag) == SA_OK) {
SaMsgQueueHandleT *key;
__cms_queue_handle_t *qhd;
key = (SaMsgQueueHandleT *)
ha_malloc(sizeof(SaMsgQueueHandleT));
qhd = (__cms_queue_handle_t *)
ha_malloc(sizeof(__cms_queue_handle_t));
memset(qhd, 0, sizeof(__cms_queue_handle_t));
qhd->queue_handle = rcmg->handle;
qhd->queue_name = *queueName;
qhd->cms_handle = hd;
*key = qhd->queue_handle;
g_hash_table_insert(hd->queue_handle_hash, key, qhd);
g_hash_table_insert(__mqhandle_hash, key, hd);
*queueHandle = *key;
}
ha_free(rcmg);
return ret;
}
SaErrorT
saMsgQueueClose(SaMsgQueueHandleT *queueHandle)
{
int ret;
client_mqueue_close_t cmg;
client_header_t *rcmg;
__cms_handle_t *hd = NULL;
__cms_queue_handle_t *qhd;
SaNameT * qname;
gpointer origkey, orighd;
if (g_hash_table_lookup_extended(__mqhandle_hash, queueHandle,
&origkey, &orighd)) {
hd = (__cms_handle_t *) orighd;
};
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, queueHandle ? *queueHandle : -1);
return SA_ERR_BAD_HANDLE;
}
qhd = g_hash_table_lookup(hd->queue_handle_hash, queueHandle);
if (!qhd) {
cl_log(LOG_ERR, "%s: Cannot find handlle [%d]"
, __FUNCTION__, queueHandle ? *queueHandle : -1);
return SA_ERR_BAD_HANDLE;
}
qname = &(qhd->queue_name);
cmg.header.type = CMS_QUEUE_CLOSE;
cmg.header.name = *qname;
cmg.handle = *queueHandle;
cmg.silent = FALSE;
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
ret = wait_for_msg(hd, CMS_QUEUE_CLOSE, qname, &rcmg, SA_TIME_END);
if (ret != SA_OK)
return ret;
if ((ret = rcmg->flag) == SA_OK) {
g_hash_table_remove(hd->queue_handle_hash, queueHandle);
g_hash_table_remove(__mqhandle_hash, queueHandle);
ha_free(origkey);
ha_free(qhd);
/* TODO: free the queue msgs. */
}
ha_free((client_mqueue_close_t *) rcmg);
return ret;
}
static void
lookup_queuehandle(gpointer key, gpointer value, gpointer user_data)
{
char * qname;
__cms_queue_handle_t *qhd = (__cms_queue_handle_t *)value;
char * name = (char *)user_data;
SaMsgQueueHandleT *queueHandle = (SaMsgQueueHandleT *)key;
qname = saname2str(qhd->queue_name);
if (!strcmp(qname, name)) {
g_hash_table_remove(qhd->cms_handle->queue_handle_hash, key);
}
g_hash_table_remove(__mqhandle_hash, queueHandle);
}
SaErrorT
saMsgQueueUnlink(SaMsgHandleT *msgHandle, const SaNameT *queueName)
{
int ret;
char * name;
client_mqueue_unlink_t cmg;
client_header_t *rcmg;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (bad_saname(queueName))
return SA_ERR_INVALID_PARAM;
cmg.header.type = CMS_QUEUE_UNLINK;
cmg.header.name = *queueName;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
ret = wait_for_msg(hd, CMS_QUEUE_UNLINK, queueName, &rcmg,
SA_TIME_END);
if (ret != SA_OK)
return ret;
/*
* remove from the mq from queue_handle_hash if possible
*/
name = saname2str(*queueName);
g_hash_table_foreach(hd->queue_handle_hash, lookup_queuehandle, name);
ret = rcmg->flag;
ha_free((client_mqueue_unlink_t *) rcmg);
ha_free(name);
return ret;
}
SaErrorT
saMsgQueueStatusGet(SaMsgHandleT *msgHandle, const SaNameT *queueName
, SaMsgQueueStatusT *queueStatus)
{
int ret;
client_mqueue_status_t cmg;
client_mqueue_status_t *rcmg;
client_header_t * reply;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (bad_saname(queueName))
return SA_ERR_INVALID_PARAM;
cmg.header.type = CMS_QUEUE_STATUS;
cmg.header.name = *queueName;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
ret = wait_for_msg(hd, CMS_QUEUE_STATUS, queueName, &reply,
SA_TIME_END);
if (ret != SA_OK)
return ret;
rcmg = (client_mqueue_status_t *) reply;
ret = reply->flag;
if (ret == SA_OK)
*queueStatus = rcmg->qstatus;
ha_free((client_mqueue_status_t *) reply);
return ret;
}
SaErrorT
saMsgMessageSend(const SaMsgHandleT *msgHandle,
const SaNameT *destination,
const SaMsgMessageT *message,
SaMsgAckFlagsT ackFlags,
SaTimeT timeout)
{
client_message_t *cmg;
client_header_t *rcmg;
client_message_ack_t * ack;
int ret;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (message->priority > SA_MSG_MESSAGE_LOWEST_PRIORITY
|| !(ackFlags & SA_MSG_MESSAGE_DELIVERED_ACK))
return SA_ERR_INVALID_PARAM;
if (ackFlags & ~SA_MSG_MESSAGE_DELIVERED_ACK)
return SA_ERR_BAD_FLAGS;
cmg = (client_message_t *)
ha_malloc(sizeof(client_message_t) + message->size);
if (!cmg)
return SA_ERR_NO_MEMORY;
cmg->header.type = CMS_MSG_SEND;
cmg->header.name = *destination;
cmg->msg = *message;
cmg->invocation = 0;
cmg->data = cmg + 1;
memcpy(cmg->data, message->data, message->size);
cmg->ack = SA_MSG_MESSAGE_DELIVERED_ACK; /* according to the spec */
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
if (ackFlags & SA_MSG_MESSAGE_DELIVERED_ACK
&& !(hd->callbacks).saMsgMessageDeliveredCallback)
return SA_ERR_INIT;
ret = cmsclient_message_send(hd,
sizeof(client_message_t) + message->size, cmg);
ha_free(cmg);
while (1) {
ret = wait_for_msg(hd, CMS_MSG_ACK | CMS_MSG_SEND,
destination, &rcmg, timeout);
if (ret != SA_OK)
return ret;
ret = rcmg->flag;
ack = (client_message_ack_t *) rcmg;
if (rcmg->type == CMS_MSG_SEND) {
ha_free((client_message_t *)rcmg);
return ret;
}
/*
* CMS_MSG_SEND is a blocking call, so we can only
* have one client waiting for it. Thus when we get
* an ACK that is for the request type CMS_MSG_SEND,
* we know this is the ACK we are waiting for.
*/
dprintf("type is %d\n", ack->send_type);
if (ack->send_type == CMS_MSG_SEND) {
ha_free((client_message_t *) rcmg);
return ret;
} else {
enqueue_dispatch_msg(hd, rcmg);
}
}
}
SaErrorT
saMsgMessageSendAsync(const SaMsgHandleT *msgHandle,
SaInvocationT invocation,
const SaNameT *destination,
const SaMsgMessageT *message,
SaMsgAckFlagsT ackFlags)
{
client_message_t *cmg;
int ret;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
cmg = (client_message_t *)
ha_malloc(sizeof(client_message_t) + message->size);
if (!cmg)
return SA_ERR_NO_MEMORY;
cmg->header.type = CMS_MSG_SEND_ASYNC;
cmg->header.name = *destination;
cmg->msg = *message;
cmg->invocation = invocation;
cmg->data = (char *)cmg + sizeof(client_message_t);
memcpy(cmg->data, message->data, message->size);
cmg->ack = ackFlags;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
ret = cmsclient_message_send(hd,
sizeof(client_message_t) + message->size, cmg);
ha_free(cmg);
return ret == IPC_OK ? SA_OK : SA_ERR_LIBRARY;
}
static int
request_for_message(__cms_handle_t * hd, const SaNameT * name)
{
client_header_t request_msg;
request_msg.type = CMS_MSG_REQUEST;
request_msg.name = *name;
return cmsclient_message_send(hd, sizeof(request_msg), &request_msg);
}
SaErrorT
saMsgMessageGet(const SaMsgQueueHandleT *queueHandle,
SaMsgMessageT *message,
SaMsgMessageInfoT *messageInfo,
SaTimeT timeout)
{
int ret;
SaErrorT error = SA_OK;
client_message_t * cmg;
client_header_t *rcmg;
client_header_t request_msg;
__cms_handle_t *hd = GET_MQ_HANDLE(queueHandle);
__cms_queue_handle_t *qhd;
SaNameT * qname;
int freecmg = 0;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by queueHandle [%d]"
, __FUNCTION__, queueHandle ? *queueHandle : -1);
return SA_ERR_BAD_HANDLE;
}
qhd = g_hash_table_lookup(hd->queue_handle_hash, queueHandle);
assert(qhd != NULL);
qname = &(qhd->queue_name);
/*
* request a message from daemon
*/
request_msg.type = CMS_MSG_REQUEST;
request_msg.name = *qname;
cmsclient_message_send(hd, sizeof(request_msg), &request_msg);
wait_again:
ret = wait_for_msg(hd, CMS_MSG_NOTIFY |CMS_MSG_GET, qname, &rcmg,
timeout);
if (rcmg->type == CMS_MSG_NOTIFY) {
dprintf("Received CMS_MSG_NOTIFY\n");
request_for_message(hd, qname);
goto wait_again;
}
if (ret != SA_OK) {
cl_log(LOG_ERR, "wait_for_msg error [%d]", ret);
return ret;
}
cmg = (client_message_t *)rcmg;
cmg->data = (void *)((char *)cmg + sizeof(client_message_t));
dprintf("message.data is [%s]\n", (char *)cmg->data);
if (message->size < cmg->msg.size)
error = SA_ERR_NO_SPACE;
message->size = (cmg->msg).size;
if (message->data) {
memcpy(message->data, cmg->data,
(cmg->msg.size > message->size ?
message->size : cmg->msg.size));
freecmg = 1;
} else {
message->data = cmg->data;
}
message->type = cmg->msg.type;
message->version = cmg->msg.version;
message->priority = cmg->msg.priority;
if (freecmg)
ha_free(cmg);
/* TODO: message info */
return error;
}
SaErrorT
saMsgMessageReceivedGet(const SaMsgQueueHandleT *queueHandle,
const SaMsgMessageHandleT *messageHandle,
SaMsgMessageT *message,
SaMsgMessageInfoT *messageInfo)
{
return SA_ERR_NOT_SUPPORTED;
}
SaErrorT
saMsgMessageCancel(const SaMsgQueueHandleT *queueHandle)
{
return SA_ERR_NOT_SUPPORTED;
}
SaErrorT
saMsgSelectionObjectGet(const SaMsgHandleT *msgHandle,
SaSelectionObjectT *selectionObject)
{
int ret;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
dprintf("hd->backup_fd is [%d]\n", hd->backup_fd);
ret = hd->backup_fd >= 0 ? hd->active_fd :
hd->ch->ops->get_recv_select_fd(hd->ch);
if (ret < 0)
return SA_ERR_LIBRARY;
*selectionObject = ret;
return SA_OK;
}
SaErrorT
saMsgDispatch(const SaMsgHandleT *msgHandle, SaDispatchFlagsT dispatchFlags)
{
int ret;
client_header_t *msg;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
switch (dispatchFlags) {
case SA_DISPATCH_ONE:
read_and_queue_ipc_msg(hd);
if ((msg = dequeue_dispatch_msg(&hd->dispatch_queue)) == NULL) {
cl_log(LOG_ERR, "%s: dequeue_dispatch_msg got NULL"
, __FUNCTION__);
return SA_OK;
}
ret = dispatch_msg(hd, msg);
if (g_list_length(hd->dispatch_queue))
active_poll(hd);
if (ret != HA_OK)
return SA_ERR_LIBRARY;
break;
case SA_DISPATCH_ALL:
read_and_queue_ipc_msg(hd);
do {
if ((msg = dequeue_dispatch_msg(&hd->dispatch_queue))
!= NULL) {
dispatch_msg(hd, msg);
}
} while (g_list_length(hd->dispatch_queue));
break;
case SA_DISPATCH_BLOCKING:
break;
default:
cl_log(LOG_ERR, "%s: wrong dispatchFlags [%d]",
__FUNCTION__, dispatchFlags);
return SA_ERR_INVALID_PARAM;
}
return SA_OK;
}
SaErrorT
saMsgQueueOpenAsync(const SaMsgHandleT *msgHandle,
SaInvocationT invocation,
const SaNameT *queueName,
const SaMsgQueueCreationAttributesT *creationAttributes,
SaMsgQueueOpenFlagsT openFlags)
{
client_mqueue_open_t cmg;
int ret;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (!openFlags)
return SA_ERR_BAD_FLAGS;
if (bad_saname(queueName) || !msgHandle
|| (!creationAttributes && !(openFlags & SA_MSG_QUEUE_OPEN_ONLY))
|| (creationAttributes && (openFlags & SA_MSG_QUEUE_OPEN_ONLY)))
return SA_ERR_INVALID_PARAM;
cmg.header.type = CMS_QUEUE_OPEN_ASYNC;
cmg.header.name = *queueName;
if (creationAttributes) {
if ((creationAttributes->creationFlags != 0)
&& creationAttributes->creationFlags !=
SA_MSG_QUEUE_PERSISTENT)
return SA_ERR_BAD_FLAGS;
cmg.attr = *creationAttributes;
}
cmg.openflag = openFlags;
cmg.invocation = invocation;
cmg.policy = 0;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
if (openFlags & SA_MSG_QUEUE_RECEIVE_CALLBACK
&& !(hd->callbacks).saMsgQueueOpenCallback
&& !(hd->callbacks).saMsgMessageReceivedCallback)
return SA_ERR_INIT;
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
if (ret != IPC_OK)
cl_log(LOG_ERR, "%s: cmsclient_message_send failed, rc = %d"
, __FUNCTION__, ret);
return ret == IPC_OK ? SA_OK : SA_ERR_LIBRARY;
}
SaErrorT
saMsgQueueGroupCreate(SaMsgHandleT *msgHandle,
const SaNameT *queueGroupName,
SaMsgQueueGroupPolicyT queueGroupPolicy)
{
int ret;
client_mqueue_open_t cmg;
client_mqueue_open_t *rcmg;
client_header_t * reply;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
cmg.header.type = CMS_QUEUEGROUP_CREATE;
cmg.header.name = *queueGroupName;
cmg.invocation = 0;
if (queueGroupPolicy != SA_MSG_QUEUE_GROUP_ROUND_ROBIN)
return SA_ERR_INVALID_PARAM;
cmg.policy = queueGroupPolicy;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
if (ret != IPC_OK) {
cl_log(LOG_ERR, "%s: cmsclient_message_send returns %d"
, __FUNCTION__,ret);
return SA_ERR_LIBRARY;
}
ret = wait_for_msg(hd, CMS_QUEUEGROUP_CREATE, queueGroupName, &reply,
SA_TIME_END);
if (ret != SA_OK)
return ret;
rcmg = (client_mqueue_open_t *) reply;
if ((rcmg->header).flag == SA_OK) {
SaMsgQueueHandleT *key;
key = (SaMsgQueueHandleT *)ha_malloc(sizeof(SaMsgQueueHandleT));
*key = rcmg->handle;
g_hash_table_insert(__mqhandle_hash, key, hd);
}
ret = (rcmg->header).flag;
ha_free(rcmg);
return ret;
}
SaErrorT
saMsgQueueGroupDelete(SaMsgHandleT *msgHandle, const SaNameT *queueGroupName)
{
/* TODO: we should remove the key that's in the __mqhandle_hash
* as well
*/
return saMsgQueueUnlink(msgHandle, queueGroupName);
}
SaErrorT
saMsgQueueGroupInsert(SaMsgHandleT *msgHandle,
const SaNameT *queueGroupName,
const SaNameT *queueName)
{
int ret;
client_header_t * rcmg;
client_mqgroup_ops_t cmg;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (bad_saname(queueName))
return SA_ERR_INVALID_PARAM;
cmg.header.type = CMS_QUEUEGROUP_INSERT;
cmg.header.name = *queueName;
cmg.qgname = *queueGroupName;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
if (ret != IPC_OK) {
cl_log(LOG_ERR, "%s: cmsclient_message_send returns %d"
, __FUNCTION__, ret);
return SA_ERR_LIBRARY;
}
ret = wait_for_msg(hd, CMS_QUEUEGROUP_INSERT, queueName, &rcmg,
SA_TIME_END);
if (ret != SA_OK)
return ret;
ret = rcmg->flag;
ha_free(rcmg);
return ret;
}
SaErrorT
saMsgQueueGroupRemove(SaMsgHandleT *msgHandle,
const SaNameT *queueGroupName,
const SaNameT *queueName)
{
int ret;
client_header_t * rcmg;
client_mqgroup_ops_t cmg;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (bad_saname(queueName))
return SA_ERR_INVALID_PARAM;
cmg.header.type = CMS_QUEUEGROUP_REMOVE;
cmg.header.name = *queueName;
cmg.qgname = *queueGroupName;
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
if (ret != IPC_OK) {
cl_log(LOG_ERR, "%s: cmsclient_message_send returns %d"
, __FUNCTION__, ret);
return SA_ERR_LIBRARY;
}
ret = wait_for_msg(hd, CMS_QUEUEGROUP_REMOVE, queueName, &rcmg,
SA_TIME_END);
if (ret != SA_OK)
return ret;
ret = rcmg->flag;
ha_free((client_mqueue_unlink_t *) rcmg);
return ret;
}
SaErrorT
saMsgQueueGroupTrackStart(const SaMsgHandleT *msgHandle,
const SaNameT *queueGroupName,
SaUint8T trackFlags,
SaMsgQueueGroupNotificationT *notificationBuffer,
SaUint32T numberOfItems)
{
int ret;
client_mqgroup_mem_t cmg;
client_header_t * rcmg;
client_mqgroup_notify_t * rmsg;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
if (!(hd->callbacks).saMsgQueueGroupTrackCallback)
return SA_ERR_INIT;
if ((trackFlags & SA_TRACK_CHANGES)
&& (trackFlags & SA_TRACK_CHANGES_ONLY))
return SA_ERR_BAD_FLAGS;
/* tell server we care about the membership information */
cmg.header.type = CMS_QUEUEGROUP_TRACK_START;
cmg.header.name = *queueGroupName;
cmg.group_name = *queueGroupName;
cmg.flag = trackFlags;
ret = cmsclient_message_send(hd, sizeof(cmg), &cmg);
if (ret != IPC_OK) {
cl_log(LOG_ERR, "%s: cmsclient_message_send returns %d"
, __FUNCTION__, ret);
return SA_ERR_LIBRARY;
}
ret = wait_for_msg(hd, CMS_QUEUEGROUP_TRACK_START, queueGroupName,
&rcmg, SA_TIME_END);
if (ret != SA_OK)
return ret;
if ((ret = rcmg->flag) != SA_OK)
return ret;
rmsg = (client_mqgroup_notify_t *)rcmg;
if ((trackFlags & SA_TRACK_CHANGES)
|| (trackFlags & SA_TRACK_CHANGES_ONLY)) {
/*
* Track membership changes with callbacks.
*/
__mqgroup_track_t * track;
char * name;
name = (char *) ha_malloc(queueGroupName->length + 1);
strncpy(name, queueGroupName->value, queueGroupName->length);
name[queueGroupName->length] = '\0';
track = (__mqgroup_track_t *)
ha_malloc(sizeof(__mqgroup_track_t));
track->name = queueGroupName;
track->flag = trackFlags & ~SA_TRACK_CURRENT;
g_hash_table_insert(__group_tracking_hash, name, track);
}
if (trackFlags & SA_TRACK_CURRENT) {
rmsg->data = (char *)rmsg + sizeof(client_mqgroup_notify_t);
if (!notificationBuffer) {
/*
* Client wants saMsgQueueGroupTrackCallback.
*/
goto exit;
}
}
exit:
ha_free(rcmg);
return ret;
}
SaErrorT
saMsgQueueGroupTrackStop(const SaMsgHandleT *msgHandle,
const SaNameT *queueGroupName)
{
char * name;
gpointer key, track;
__cms_handle_t *hd = GET_CMS_HANDLE(msgHandle);
if (hd == NULL) {
cl_log(LOG_ERR, "%s: Cannot find hd by handlle [%d]"
, __FUNCTION__, msgHandle ? *msgHandle : -1);
return SA_ERR_BAD_HANDLE;
}
if (bad_saname(queueGroupName))
return SA_ERR_INVALID_PARAM;
name = (char *) ha_malloc(queueGroupName->length + 1);
strncpy(name, queueGroupName->value, queueGroupName->length);
name[queueGroupName->length] = '\0';
if (g_hash_table_lookup_extended(__group_tracking_hash, name, &key
, &track) == TRUE) {
g_hash_table_remove(__group_tracking_hash, key);
ha_free(key);
} else
return SA_ERR_NOT_EXIST;
ha_free(name);
return SA_OK;
}
syntax highlighted by Code2HTML, v. 0.9.1