/*
Copyright (c) 2007-2009 FastMQ Inc.
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the Lesser GNU General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
0MQ 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
Lesser GNU General Public License for more details.
You should have received a copy of the Lesser GNU General Public License
along with this program. If not, see .
*/
#include "../include/zs.h"
#include
#include
#include
#include "i_api.hpp"
#include "err.hpp"
#include "dispatcher.hpp"
#include "msg.hpp"
int zs_msg_init (zs_msg *msg_)
{
msg_->content = (zs_msg_content*) ZS_VSM;
msg_->vsm_size = 0;
return 0;
}
int zs_msg_init_size (zs_msg *msg_, size_t size_)
{
if (size_ <= ZS_MAX_VSM_SIZE) {
msg_->content = (zs_msg_content*) ZS_VSM;
msg_->vsm_size = (uint16_t) size_;
}
else {
msg_->content = (zs_msg_content*) malloc (sizeof (zs_msg_content) +
size_);
if (!msg_->content) {
errno = ENOMEM;
return -1;
}
msg_->shared = 0;
msg_->content->data = (void*) (msg_->content + 1);
msg_->content->size = size_;
msg_->content->ffn = NULL;
new (&msg_->content->refcnt) zs::atomic_counter_t ();
}
return 0;
}
int zs_msg_init_data (zs_msg *msg_, void *data_, size_t size_, zs_free_fn *ffn_)
{
msg_->shared = 0;
msg_->content = (zs_msg_content*) malloc (sizeof (zs_msg_content));
zs_assert (msg_->content);
msg_->content->data = data_;
msg_->content->size = size_;
msg_->content->ffn = ffn_;
new (&msg_->content->refcnt) zs::atomic_counter_t ();
return 0;
}
int zs_msg_close (zs_msg *msg_)
{
// For VSMs and delimiters there are no resources to free
if (msg_->content == (zs_msg_content*) ZS_DELIMITER ||
msg_->content == (zs_msg_content*) ZS_VSM ||
msg_->content == (zs_msg_content*) ZS_GAP)
return 0;
// If the content is not shared, or if it is shared and the reference
// count has dropped to zero, deallocate it.
if (!msg_->shared || !msg_->content->refcnt.sub (1)) {
// We used "placement new" operator to initialize the reference
// counter so we call its destructor now.
msg_->content->refcnt.~atomic_counter_t ();
if (msg_->content->ffn)
msg_->content->ffn (msg_->content->data);
free (msg_->content);
}
return 0;
}
int zs_msg_move (zs_msg *dest_, zs_msg *src_)
{
zs_msg_close (dest_);
*dest_ = *src_;
zs_msg_init (src_);
return 0;
}
int zs_msg_copy (zs_msg *dest_, zs_msg *src_)
{
zs_msg_close (dest_);
// VSMs and delimiters require no special handling.
if (src_->content !=
(zs_msg_content*) ZS_DELIMITER &&
src_->content != (zs_msg_content*) ZS_VSM &&
src_->content != (zs_msg_content*) ZS_GAP) {
// One reference is added to shared messages. Non-shared messages
// are turned into shared messages and reference count is set to 2.
if (src_->shared)
src_->content->refcnt.add (1);
else {
src_->shared = true;
src_->content->refcnt.set (2);
}
}
*dest_ = *src_;
return 0;
}
void *zs_msg_data (zs_msg *msg_)
{
if (msg_->content == (zs_msg_content*) ZS_VSM)
return msg_->vsm_data;
if (msg_->content ==
(zs_msg_content*) ZS_DELIMITER ||
msg_->content == (zs_msg_content*) ZS_GAP)
return NULL;
return msg_->content->data;
}
size_t zs_msg_size (zs_msg *msg_)
{
if (msg_->content == (zs_msg_content*) ZS_VSM)
return msg_->vsm_size;
if (msg_->content ==
(zs_msg_content*) ZS_DELIMITER ||
msg_->content == (zs_msg_content*) ZS_GAP)
return 0;
return msg_->content->size;
}
int zs_msg_type (zs_msg *msg_)
{
// If it's a genuine message, return 0.
if (msg_->content >= (zs_msg_content*) ZS_VSM)
return 0;
// Trick the compiler to believe that content is an integer.
unsigned char *offset = 0;
return (((const unsigned char*) msg_->content) - offset);
}
void *zs_init (int app_threads_, int io_threads_)
{
// There should be at least a single thread managed by the dispatcher.
if (app_threads_ < 0 || io_threads_ < 0 ||
app_threads_ + io_threads_ == 0) {
errno = EINVAL;
return NULL;
}
zs::dispatcher_t *dispatcher =
new zs::dispatcher_t (app_threads_, io_threads_);
zs_assert (dispatcher);
return (void*) dispatcher;
}
int zs_term (void *context_)
{
((zs::dispatcher_t*) context_)->shutdown ();
return 0;
}
void *zs_socket (void *context_, int type_)
{
return (void*) (((zs::dispatcher_t*) context_)->create_socket (type_));
}
int zs_close (void *s_)
{
((zs::i_api*) s_)->close ();
return 0;
}
int zs_bind (void *s_, const char *addr_, zs_opts *opts_)
{
return (((zs::i_api*) s_)->bind (addr_, opts_));
}
int zs_connect (void *s_, const char *addr_, zs_opts *opts_)
{
return (((zs::i_api*) s_)->connect (addr_, opts_));
}
int zs_subscribe (void *s_, const char *criteria_)
{
return (((zs::i_api*) s_)->subscribe (criteria_));
}
int zs_send (void *s_, zs_msg *msg_, int flags_)
{
return (((zs::i_api*) s_)->send (msg_, flags_));
}
int zs_flush (void *s_)
{
return (((zs::i_api*) s_)->flush ());
}
int zs_recv (void *s_, zs_msg *msg_, int flags_)
{
return (((zs::i_api*) s_)->recv (msg_, flags_));
}