Page MenuHomePhabricator (Chris)
Files F628547

amqp_exchange.c
No OneTemporary
Actions

Authored By
Unknown
Size
30 KB
Referenced Files
None
Subscribers
None

amqp_exchange.c
View Options

/*
+----------------------------------------------------------------------+
| PHP Version 5 |
+----------------------------------------------------------------------+
| Copyright (c) 1997-2007 The PHP Group |
+----------------------------------------------------------------------+
| This source file is subject to version 3.01 of the PHP license, |
| that is bundled with this package in the file LICENSE, and is |
| available through the world-wide-web at the following url: |
| http://www.php.net/license/3_01.txt |
| If you did not receive a copy of the PHP license and are unable to |
| obtain it through the world-wide-web, please send a note to |
| license@php.net so we can mail you a copy immediately. |
+----------------------------------------------------------------------+
| Author: Alexandre Kalendarev akalend@mail.ru Copyright (c) 2009-2010 |
| Lead: |
| - Pieter de Zwart |
| Maintainers: |
| - Brad Rodriguez |
| - Jonathan Tansavatdi |
+----------------------------------------------------------------------+
*/
/* $Id: amqp_exchange.c 327551 2012-09-09 03:49:34Z pdezwart $ */
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "php.h"
#include "php_ini.h"
#include "ext/standard/info.h"
#include "zend_exceptions.h"
#ifdef PHP_WIN32
# include "win32/php_stdint.h"
# include "win32/signal.h"
#else
# include <signal.h>
# include <stdint.h>
#endif
#include <amqp.h>
#include <amqp_framing.h>
#ifdef PHP_WIN32
# include "win32/unistd.h"
#else
# include <unistd.h>
#endif
#include "php_amqp.h"
#include "amqp_connection.h"
#include "amqp_channel.h"
#include "amqp_exchange.h"
zend_class_entry *amqp_exchange_class_entry;
#define this_ce amqp_exchange_class_entry
/* {{{ proto AMQPExchange::__construct(AMQPChannel channel);
create Exchange */
PHP_METHOD(amqp_exchange_class, __construct)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP5to7_zval_t arguments PHP5to7_MAYBE_SET_TO_NULL;
zval *channelObj;
amqp_channel_resource *channel_resource;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "o", &channelObj) == FAILURE) {
return;
}
PHP5to7_MAYBE_INIT(arguments);
PHP5to7_ARRAY_INIT(arguments);
zend_update_property(this_ce, getThis(), ZEND_STRL("arguments"), PHP5to7_MAYBE_PTR(arguments) TSRMLS_CC);
PHP5to7_MAYBE_DESTROY(arguments);
channel_resource = PHP_AMQP_GET_CHANNEL_RESOURCE(channelObj);
PHP_AMQP_VERIFY_CHANNEL_RESOURCE(channel_resource, "Could not create exchange.");
zend_update_property(this_ce, getThis(), ZEND_STRL("channel"), channelObj TSRMLS_CC);
zend_update_property(this_ce, getThis(), ZEND_STRL("connection"), PHP_AMQP_READ_OBJ_PROP(amqp_channel_class_entry, channelObj, "connection") TSRMLS_CC);
}
/* }}} */
/* {{{ proto AMQPExchange::getName()
Get the exchange name */
PHP_METHOD(amqp_exchange_class, getName)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP_AMQP_NOPARAMS();
if (PHP_AMQP_READ_THIS_PROP_STRLEN("name") > 0) {
PHP_AMQP_RETURN_THIS_PROP("name");
} else {
/* BC */
RETURN_FALSE;
}
}
/* }}} */
/* {{{ proto AMQPExchange::setName(string name)
Set the exchange name */
PHP_METHOD(amqp_exchange_class, setName)
{
char *name = NULL;
PHP5to7_param_str_len_type_t name_len = 0;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s", &name, &name_len) == FAILURE) {
return;
}
/* Verify that the name is not null and not an empty string */
if (name_len > 255) {
zend_throw_exception(amqp_exchange_exception_class_entry, "Invalid exchange name given, must be less than 255 characters long.", 0 TSRMLS_CC);
return;
}
/* Set the exchange name */
zend_update_property_stringl(this_ce, getThis(), ZEND_STRL("name"), name, name_len TSRMLS_CC);
}
/* }}} */
/* {{{ proto AMQPExchange::getFlags()
Get the exchange parameters */
PHP_METHOD(amqp_exchange_class, getFlags)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP5to7_param_long_type_t flagBitmask = 0;
PHP_AMQP_NOPARAMS();
if (PHP_AMQP_READ_THIS_PROP_BOOL("passive")) {
flagBitmask |= AMQP_PASSIVE;
}
if (PHP_AMQP_READ_THIS_PROP_BOOL("durable")) {
flagBitmask |= AMQP_DURABLE;
}
if (PHP_AMQP_READ_THIS_PROP_BOOL("auto_delete")) {
flagBitmask |= AMQP_AUTODELETE;
}
if (PHP_AMQP_READ_THIS_PROP_BOOL("internal")) {
flagBitmask |= AMQP_INTERNAL;
}
RETURN_LONG(flagBitmask);
}
/* }}} */
/* {{{ proto AMQPExchange::setFlags(long bitmask)
Set the exchange parameters */
PHP_METHOD(amqp_exchange_class, setFlags)
{
PHP5to7_param_long_type_t flagBitmask;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "l", &flagBitmask) == FAILURE) {
return;
}
/* Set the flags based on the bitmask we were given */
flagBitmask = flagBitmask ? flagBitmask & PHP_AMQP_EXCHANGE_FLAGS : flagBitmask;
zend_update_property_bool(this_ce, getThis(), ZEND_STRL("passive"), IS_PASSIVE(flagBitmask) TSRMLS_CC);
zend_update_property_bool(this_ce, getThis(), ZEND_STRL("durable"), IS_DURABLE(flagBitmask) TSRMLS_CC);
zend_update_property_bool(this_ce, getThis(), ZEND_STRL("auto_delete"), IS_AUTODELETE(flagBitmask) TSRMLS_CC);
zend_update_property_bool(this_ce, getThis(), ZEND_STRL("internal"), IS_INTERNAL(flagBitmask) TSRMLS_CC);
}
/* }}} */
/* {{{ proto AMQPExchange::getType()
Get the exchange type */
PHP_METHOD(amqp_exchange_class, getType)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP_AMQP_NOPARAMS();
if (PHP_AMQP_READ_THIS_PROP_STRLEN("type") > 0) {
PHP_AMQP_RETURN_THIS_PROP("type");
} else {
/* BC */
RETURN_FALSE;
}
}
/* }}} */
/* {{{ proto AMQPExchange::setType(string type)
Set the exchange type */
PHP_METHOD(amqp_exchange_class, setType)
{
char *type = NULL; PHP5to7_param_str_len_type_t type_len = 0;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s", &type, &type_len) == FAILURE) {
return;
}
zend_update_property_stringl(this_ce, getThis(), ZEND_STRL("type"), type, type_len TSRMLS_CC);
}
/* }}} */
/* {{{ proto AMQPExchange::getArgument(string key)
Get the exchange argument referenced by key */
PHP_METHOD(amqp_exchange_class, getArgument)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP5to7_zval_t *tmp = NULL;
char *key; PHP5to7_param_str_len_type_t key_len;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s", &key, &key_len) == FAILURE) {
return;
}
if (!PHP5to7_ZEND_HASH_FIND(PHP_AMQP_READ_THIS_PROP_ARR("arguments"), key, key_len + 1, tmp)) {
RETURN_FALSE;
}
RETURN_ZVAL(PHP5to7_MAYBE_DEREF(tmp), 1, 0);
}
/* }}} */
/* {{{ proto AMQPExchange::hasArgument(string key) */
PHP_METHOD(amqp_exchange_class, hasArgument)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP5to7_zval_t *tmp = NULL;
char *key; PHP5to7_param_str_len_type_t key_len;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s", &key, &key_len) == FAILURE) {
return;
}
if (!PHP5to7_ZEND_HASH_FIND(PHP_AMQP_READ_THIS_PROP_ARR("arguments"), key, (uint)(key_len + 1), tmp)) {
RETURN_FALSE;
}
RETURN_TRUE;
}
/* }}} */
/* {{{ proto AMQPExchange::getArguments
Get the exchange arguments */
PHP_METHOD(amqp_exchange_class, getArguments)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP_AMQP_NOPARAMS();
PHP_AMQP_RETURN_THIS_PROP("arguments");
}
/* }}} */
/* {{{ proto AMQPExchange::setArguments(array args)
Overwrite all exchange arguments with given args */
PHP_METHOD(amqp_exchange_class, setArguments)
{
zval *zvalArguments;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "a", &zvalArguments) == FAILURE) {
return;
}
zend_update_property(this_ce, getThis(), ZEND_STRL("arguments"), zvalArguments TSRMLS_CC);
RETURN_TRUE;
}
/* }}} */
/* {{{ proto AMQPExchange::setArgument(key, value) */
PHP_METHOD(amqp_exchange_class, setArgument)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
char *key= NULL; PHP5to7_param_str_len_type_t key_len = 0;
zval *value = NULL;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "sz",
&key, &key_len,
&value) == FAILURE) {
return;
}
switch (Z_TYPE_P(value)) {
case IS_NULL:
PHP5to7_ZEND_HASH_DEL(PHP_AMQP_READ_THIS_PROP_ARR("arguments"), key, (uint) (key_len + 1));
break;
PHP5to7_CASE_IS_BOOL:
case IS_LONG:
case IS_DOUBLE:
case IS_STRING:
PHP5to7_ZEND_HASH_ADD(PHP_AMQP_READ_THIS_PROP_ARR("arguments"), key, (uint) (key_len + 1), value, sizeof(zval *));
Z_TRY_ADDREF_P(value);
break;
default:
zend_throw_exception(amqp_exchange_exception_class_entry, "The value parameter must be of type NULL, int, double or string.", 0 TSRMLS_CC);
return;
}
RETURN_TRUE;
}
/* }}} */
/* {{{ proto AMQPExchange::declareExchange();
declare Exchange
*/
PHP_METHOD(amqp_exchange_class, declareExchange)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
amqp_channel_resource *channel_resource;
amqp_table_t *arguments;
if (zend_parse_parameters_none() == FAILURE) {
return;
}
channel_resource = PHP_AMQP_GET_CHANNEL_RESOURCE(PHP_AMQP_READ_THIS_PROP("channel"));
PHP_AMQP_VERIFY_CHANNEL_RESOURCE(channel_resource, "Could not declare exchange.");
/* Check that the exchange has a name */
if (PHP_AMQP_READ_THIS_PROP_STRLEN("name") < 1) {
zend_throw_exception(amqp_exchange_exception_class_entry, "Could not declare exchange. Exchanges must have a name.", 0 TSRMLS_CC);
return;
}
/* Check that the exchange has a name */
if (PHP_AMQP_READ_THIS_PROP_STRLEN("type") < 1) {
zend_throw_exception(amqp_exchange_exception_class_entry, "Could not declare exchange. Exchanges must have a type.", 0 TSRMLS_CC);
return;
}
arguments = convert_zval_to_amqp_table(PHP_AMQP_READ_THIS_PROP("arguments") TSRMLS_CC);
#if AMQP_VERSION_MAJOR * 100 + AMQP_VERSION_MINOR * 10 + AMQP_VERSION_PATCH > 52
amqp_exchange_declare(
channel_resource->connection_resource->connection_state,
channel_resource->channel_id,
amqp_cstring_bytes(PHP_AMQP_READ_THIS_PROP_STR("name")),
amqp_cstring_bytes(PHP_AMQP_READ_THIS_PROP_STR("type")),
PHP_AMQP_READ_THIS_PROP_BOOL("passive"),
PHP_AMQP_READ_THIS_PROP_BOOL("durable"),
PHP_AMQP_READ_THIS_PROP_BOOL("auto_delete"),
PHP_AMQP_READ_THIS_PROP_BOOL("internal"),
*arguments
);
#else
amqp_exchange_declare(
channel_resource->connection_resource->connection_state,
channel_resource->channel_id,
amqp_cstring_bytes(PHP_AMQP_READ_THIS_PROP_STR("name")),
amqp_cstring_bytes(PHP_AMQP_READ_THIS_PROP_STR("type")),
PHP_AMQP_READ_THIS_PROP_BOOL("passive"),
PHP_AMQP_READ_THIS_PROP_BOOL("durable"),
*arguments
);
#endif
amqp_rpc_reply_t res = amqp_get_rpc_reply(channel_resource->connection_resource->connection_state);
php_amqp_free_amqp_table(arguments);
/* handle any errors that occured outside of signals */
if (res.reply_type != AMQP_RESPONSE_NORMAL) {
PHP_AMQP_INIT_ERROR_MESSAGE();
php_amqp_error(res, PHP_AMQP_ERROR_MESSAGE_PTR, channel_resource->connection_resource, channel_resource TSRMLS_CC);
php_amqp_zend_throw_exception(res, amqp_exchange_exception_class_entry, PHP_AMQP_ERROR_MESSAGE, 0 TSRMLS_CC);
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
PHP_AMQP_DESTROY_ERROR_MESSAGE();
return;
}
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
RETURN_TRUE;
}
/* }}} */
/* {{{ proto AMQPExchange::delete([string name[, long params]]);
delete Exchange
*/
PHP_METHOD(amqp_exchange_class, delete)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
amqp_channel_resource *channel_resource;
char *name = NULL; PHP5to7_param_str_len_type_t name_len = 0;
PHP5to7_param_long_type_t flags = 0;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "|sl",
&name, &name_len,
&flags) == FAILURE) {
return;
}
channel_resource = PHP_AMQP_GET_CHANNEL_RESOURCE(PHP_AMQP_READ_THIS_PROP("channel"));
PHP_AMQP_VERIFY_CHANNEL_RESOURCE(channel_resource, "Could not delete exchange.");
amqp_exchange_delete(
channel_resource->connection_resource->connection_state,
channel_resource->channel_id,
amqp_cstring_bytes(name_len ? name : PHP_AMQP_READ_THIS_PROP_STR("name")),
(AMQP_IFUNUSED & flags) ? 1 : 0
);
amqp_rpc_reply_t res = amqp_get_rpc_reply(channel_resource->connection_resource->connection_state);
if (res.reply_type != AMQP_RESPONSE_NORMAL) {
PHP_AMQP_INIT_ERROR_MESSAGE();
php_amqp_error(res, PHP_AMQP_ERROR_MESSAGE_PTR, channel_resource->connection_resource, channel_resource TSRMLS_CC);
php_amqp_zend_throw_exception(res, amqp_exchange_exception_class_entry, PHP_AMQP_ERROR_MESSAGE, 0 TSRMLS_CC);
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
PHP_AMQP_DESTROY_ERROR_MESSAGE();
return;
}
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
RETURN_TRUE;
}
/* }}} */
/* {{{ proto AMQPExchange::publish(string msg, [string key, [int flags, [array headers]]]);
publish into Exchange
*/
PHP_METHOD(amqp_exchange_class, publish)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
zval *ini_arr = NULL;
PHP5to7_zval_t *tmp = NULL;
amqp_channel_resource *channel_resource;
char *key_name = NULL; PHP5to7_param_str_len_type_t key_len = 0;
char *msg = NULL; PHP5to7_param_str_len_type_t msg_len = 0;
PHP5to7_param_long_type_t flags = AMQP_NOPARAM;
#ifndef PHP_WIN32
/* Storage for previous signal handler during SIGPIPE override */
void * old_handler;
#endif
amqp_basic_properties_t props;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|sla/",
&msg, &msg_len,
&key_name, &key_len,
&flags,
&ini_arr) == FAILURE) {
return;
}
/* By default (and for BC) content type is text/plain (may be skipped at all, then set props._flags to 0) */
props.content_type = amqp_cstring_bytes("text/plain");
props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG;
props.headers.entries = 0;
{
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "content_type", sizeof("content_type"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.content_type = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_CONTENT_TYPE_FLAG;
}
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "content_encoding", sizeof("content_encoding"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.content_encoding = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_CONTENT_ENCODING_FLAG;
}
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "message_id", sizeof("message_id"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.message_id = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_MESSAGE_ID_FLAG;
}
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "user_id", sizeof("user_id"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.user_id = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_USER_ID_FLAG;
}
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "app_id", sizeof("app_id"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.app_id = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_APP_ID_FLAG;
}
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "delivery_mode", sizeof("delivery_mode"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_long(PHP5to7_MAYBE_DEREF(tmp));
props.delivery_mode = (uint8_t)Z_LVAL_P(PHP5to7_MAYBE_DEREF(tmp));
props._flags |= AMQP_BASIC_DELIVERY_MODE_FLAG;
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "priority", sizeof("priority"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_long(PHP5to7_MAYBE_DEREF(tmp));
props.priority = (uint8_t)Z_LVAL_P(PHP5to7_MAYBE_DEREF(tmp));
props._flags |= AMQP_BASIC_PRIORITY_FLAG;
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "timestamp", sizeof("timestamp"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_long(PHP5to7_MAYBE_DEREF(tmp));
props.timestamp = (uint64_t)Z_LVAL_P(PHP5to7_MAYBE_DEREF(tmp));
props._flags |= AMQP_BASIC_TIMESTAMP_FLAG;
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "expiration", sizeof("expiration"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.expiration = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_EXPIRATION_FLAG;
}
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "type", sizeof("type"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.type = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_TYPE_FLAG;
}
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "reply_to", sizeof("reply_to"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.reply_to = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_REPLY_TO_FLAG;
}
}
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF (ini_arr), "correlation_id", sizeof("correlation_id"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_string(PHP5to7_MAYBE_DEREF(tmp));
if (Z_STRLEN_P(PHP5to7_MAYBE_DEREF(tmp)) > 0) {
props.correlation_id = amqp_cstring_bytes(Z_STRVAL_P(PHP5to7_MAYBE_DEREF(tmp)));
props._flags |= AMQP_BASIC_CORRELATION_ID_FLAG;
}
}
}
amqp_table_t *headers = NULL;
if (ini_arr && PHP5to7_ZEND_HASH_FIND(HASH_OF(ini_arr), "headers", sizeof("headers"), tmp)) {
SEPARATE_ZVAL(tmp);
convert_to_array(PHP5to7_MAYBE_DEREF(tmp));
headers = convert_zval_to_amqp_table(PHP5to7_MAYBE_DEREF(tmp) TSRMLS_CC);
props._flags |= AMQP_BASIC_HEADERS_FLAG;
props.headers = *headers;
}
channel_resource = PHP_AMQP_GET_CHANNEL_RESOURCE(PHP_AMQP_READ_THIS_PROP("channel"));
PHP_AMQP_VERIFY_CHANNEL_RESOURCE(channel_resource, "Could not publish to exchange.");
#ifndef PHP_WIN32
/* Start ignoring SIGPIPE */
old_handler = signal(SIGPIPE, SIG_IGN);
#endif
zval *exchange_name = PHP_AMQP_READ_THIS_PROP("name");
/* NOTE: basic.publish is asynchronous and thus will not indicate failure if something goes wrong on the broker */
int status = amqp_basic_publish(
channel_resource->connection_resource->connection_state,
channel_resource->channel_id,
(Z_TYPE_P(exchange_name) == IS_STRING && Z_STRLEN_P(exchange_name) > 0 ? amqp_cstring_bytes(Z_STRVAL_P(exchange_name)) : amqp_empty_bytes), /* exchange */
(key_len > 0 ? amqp_cstring_bytes(key_name) : amqp_empty_bytes), /* routing key */
(AMQP_MANDATORY & flags) ? 1 : 0, /* mandatory */
(AMQP_IMMEDIATE & flags) ? 1 : 0, /* immediate */
&props,
php_amqp_long_string(msg, msg_len) /* message body */
);
if (headers) {
php_amqp_free_amqp_table(headers);
}
#ifndef PHP_WIN32
/* End ignoring of SIGPIPEs */
signal(SIGPIPE, old_handler);
#endif
if (status != AMQP_STATUS_OK) {
/* Emulate library error */
amqp_rpc_reply_t res;
res.reply_type = AMQP_RESPONSE_LIBRARY_EXCEPTION;
res.library_error = status;
PHP_AMQP_INIT_ERROR_MESSAGE();
php_amqp_error(res, PHP_AMQP_ERROR_MESSAGE_PTR, channel_resource->connection_resource, channel_resource TSRMLS_CC);
php_amqp_zend_throw_exception(res, amqp_queue_exception_class_entry, PHP_AMQP_ERROR_MESSAGE, 0 TSRMLS_CC);
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
PHP_AMQP_DESTROY_ERROR_MESSAGE();
return;
}
RETURN_TRUE;
}
/* }}} */
/* {{{ proto int exchange::bind(string srcExchangeName[, string routingKey, array arguments]);
bind exchange to exchange by routing key
*/
PHP_METHOD(amqp_exchange_class, bind)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
zval *zvalArguments = NULL;
amqp_channel_resource *channel_resource;
char *src_name; PHP5to7_param_str_len_type_t src_name_len = 0;
char *keyname; PHP5to7_param_str_len_type_t keyname_len = 0;
amqp_table_t *arguments = NULL;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|sa",
&src_name, &src_name_len,
&keyname, &keyname_len,
&zvalArguments) == FAILURE) {
return;
}
channel_resource = PHP_AMQP_GET_CHANNEL_RESOURCE(PHP_AMQP_READ_THIS_PROP("channel"));
PHP_AMQP_VERIFY_CHANNEL_RESOURCE(channel_resource, "Could not bind to exchange.");
if (zvalArguments) {
arguments = convert_zval_to_amqp_table(zvalArguments TSRMLS_CC);
}
amqp_exchange_bind(
channel_resource->connection_resource->connection_state,
channel_resource->channel_id,
amqp_cstring_bytes(PHP_AMQP_READ_THIS_PROP_STR("name")),
(src_name_len > 0 ? amqp_cstring_bytes(src_name) : amqp_empty_bytes),
(keyname_len > 0 ? amqp_cstring_bytes(keyname) : amqp_empty_bytes),
(arguments ? *arguments : amqp_empty_table)
);
if (arguments) {
php_amqp_free_amqp_table(arguments);
}
amqp_rpc_reply_t res = amqp_get_rpc_reply(channel_resource->connection_resource->connection_state);
if (res.reply_type != AMQP_RESPONSE_NORMAL) {
PHP_AMQP_INIT_ERROR_MESSAGE();
php_amqp_error(res, PHP_AMQP_ERROR_MESSAGE_PTR, channel_resource->connection_resource, channel_resource TSRMLS_CC);
php_amqp_zend_throw_exception(res, amqp_exchange_exception_class_entry, PHP_AMQP_ERROR_MESSAGE, 0 TSRMLS_CC);
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
PHP_AMQP_DESTROY_ERROR_MESSAGE();
return;
}
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
RETURN_TRUE;
}
/* }}} */
/* {{{ proto int exchange::unbind(string srcExchangeName[, string routingKey, array arguments]);
remove exchange to exchange binding by routing key
*/
PHP_METHOD(amqp_exchange_class, unbind)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
zval *zvalArguments = NULL;
amqp_channel_resource *channel_resource;
char *src_name; PHP5to7_param_str_len_type_t src_name_len = 0;
char *keyname; PHP5to7_param_str_len_type_t keyname_len = 0;
amqp_table_t *arguments = NULL;
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "s|sa",
&src_name, &src_name_len,
&keyname, &keyname_len,
&zvalArguments) == FAILURE) {
return;
}
channel_resource = PHP_AMQP_GET_CHANNEL_RESOURCE(PHP_AMQP_READ_THIS_PROP("channel"));
PHP_AMQP_VERIFY_CHANNEL_RESOURCE(channel_resource, "Could not unbind from exchange.");
if (zvalArguments) {
arguments = convert_zval_to_amqp_table(zvalArguments TSRMLS_CC);
}
amqp_exchange_unbind(
channel_resource->connection_resource->connection_state,
channel_resource->channel_id,
amqp_cstring_bytes(PHP_AMQP_READ_THIS_PROP_STR("name")),
(src_name_len > 0 ? amqp_cstring_bytes(src_name) : amqp_empty_bytes),
(keyname_len > 0 ? amqp_cstring_bytes(keyname) : amqp_empty_bytes),
(arguments ? *arguments : amqp_empty_table)
);
if (arguments) {
php_amqp_free_amqp_table(arguments);
}
amqp_rpc_reply_t res = amqp_get_rpc_reply(channel_resource->connection_resource->connection_state);
if (res.reply_type != AMQP_RESPONSE_NORMAL) {
PHP_AMQP_INIT_ERROR_MESSAGE();
php_amqp_error(res, PHP_AMQP_ERROR_MESSAGE_PTR, channel_resource->connection_resource, channel_resource TSRMLS_CC);
php_amqp_zend_throw_exception(res, amqp_exchange_exception_class_entry, PHP_AMQP_ERROR_MESSAGE, 0 TSRMLS_CC);
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
PHP_AMQP_DESTROY_ERROR_MESSAGE();
return;
}
php_amqp_maybe_release_buffers_on_channel(channel_resource->connection_resource, channel_resource);
RETURN_TRUE;
}
/* }}} */
/* {{{ proto AMQPExchange::getChannel()
Get the AMQPChannel object in use */
PHP_METHOD(amqp_exchange_class, getChannel)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP_AMQP_NOPARAMS();
PHP_AMQP_RETURN_THIS_PROP("channel");
}
/* }}} */
/* {{{ proto AMQPExchange::getConnection()
Get the AMQPConnection object in use */
PHP_METHOD(amqp_exchange_class, getConnection)
{
PHP5to7_READ_PROP_RV_PARAM_DECL;
PHP_AMQP_NOPARAMS();
PHP_AMQP_RETURN_THIS_PROP("connection");
}
/* }}} */
/* amqp_exchange ARG_INFO definition */
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class__construct, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 1)
ZEND_ARG_OBJ_INFO(0, amqp_channel, AMQPChannel, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_getName, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_setName, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 1)
ZEND_ARG_INFO(0, exchange_name)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_getFlags, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_setFlags, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 1)
ZEND_ARG_INFO(0, flags)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_getType, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_setType, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 1)
ZEND_ARG_INFO(0, exchange_type)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_getArgument, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 1)
ZEND_ARG_INFO(0, argument)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_hasArgument, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 1)
ZEND_ARG_INFO(0, argument)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_getArguments, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_setArgument, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 2)
ZEND_ARG_INFO(0, key)
ZEND_ARG_INFO(0, value)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_setArguments, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 1)
ZEND_ARG_ARRAY_INFO(0, arguments, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_declareExchange, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_bind, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 2)
ZEND_ARG_INFO(0, exchange_name)
ZEND_ARG_INFO(0, routing_key)
ZEND_ARG_INFO(0, flags)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_unbind, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 2)
ZEND_ARG_INFO(0, exchange_name)
ZEND_ARG_INFO(0, routing_key)
ZEND_ARG_INFO(0, flags)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_delete, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 0)
ZEND_ARG_INFO(0, exchange_name)
ZEND_ARG_INFO(0, flags)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_publish, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 1)
ZEND_ARG_INFO(0, message)
ZEND_ARG_INFO(0, routing_key)
ZEND_ARG_INFO(0, flags)
ZEND_ARG_ARRAY_INFO(0, headers, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_getChannel, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_amqp_exchange_class_getConnection, ZEND_SEND_BY_VAL, ZEND_RETURN_VALUE, 0)
ZEND_END_ARG_INFO()
zend_function_entry amqp_exchange_class_functions[] = {
PHP_ME(amqp_exchange_class, __construct, arginfo_amqp_exchange_class__construct, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, getName, arginfo_amqp_exchange_class_getName, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, setName, arginfo_amqp_exchange_class_setName, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, getFlags, arginfo_amqp_exchange_class_getFlags, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, setFlags, arginfo_amqp_exchange_class_setFlags, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, getType, arginfo_amqp_exchange_class_getType, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, setType, arginfo_amqp_exchange_class_setType, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, getArgument, arginfo_amqp_exchange_class_getArgument, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, getArguments, arginfo_amqp_exchange_class_getArguments, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, setArgument, arginfo_amqp_exchange_class_setArgument, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, setArguments, arginfo_amqp_exchange_class_setArguments, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, hasArgument, arginfo_amqp_exchange_class_hasArgument, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, declareExchange,arginfo_amqp_exchange_class_declareExchange,ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, bind, arginfo_amqp_exchange_class_bind, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, unbind, arginfo_amqp_exchange_class_unbind, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, delete, arginfo_amqp_exchange_class_delete, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, publish, arginfo_amqp_exchange_class_publish, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, getChannel, arginfo_amqp_exchange_class_getChannel, ZEND_ACC_PUBLIC)
PHP_ME(amqp_exchange_class, getConnection, arginfo_amqp_exchange_class_getConnection, ZEND_ACC_PUBLIC)
PHP_MALIAS(amqp_exchange_class, declare, declareExchange, arginfo_amqp_exchange_class_declareExchange, ZEND_ACC_PUBLIC | ZEND_ACC_DEPRECATED)
{NULL, NULL, NULL} /* Must be the last line in amqp_functions[] */
};
PHP_MINIT_FUNCTION(amqp_exchange)
{
zend_class_entry ce;
INIT_CLASS_ENTRY(ce, "AMQPExchange", amqp_exchange_class_functions);
this_ce = zend_register_internal_class(&ce TSRMLS_CC);
zend_declare_property_null(this_ce, ZEND_STRL("connection"), ZEND_ACC_PRIVATE TSRMLS_CC);
zend_declare_property_null(this_ce, ZEND_STRL("channel"), ZEND_ACC_PRIVATE TSRMLS_CC);
zend_declare_property_null(this_ce, ZEND_STRL("name"), ZEND_ACC_PRIVATE TSRMLS_CC);
zend_declare_property_null(this_ce, ZEND_STRL("type"), ZEND_ACC_PRIVATE TSRMLS_CC);
zend_declare_property_bool(this_ce, ZEND_STRL("passive"), 0, ZEND_ACC_PRIVATE TSRMLS_CC);
zend_declare_property_bool(this_ce, ZEND_STRL("durable"), 0, ZEND_ACC_PRIVATE TSRMLS_CC);
zend_declare_property_bool(this_ce, ZEND_STRL("auto_delete"), 0, ZEND_ACC_PRIVATE TSRMLS_CC);
zend_declare_property_bool(this_ce, ZEND_STRL("internal"), 0, ZEND_ACC_PRIVATE TSRMLS_CC);
zend_declare_property_null(this_ce, ZEND_STRL("arguments"), ZEND_ACC_PRIVATE TSRMLS_CC);
return SUCCESS;
}
/*
*Local variables:
*tab-width: 4
*c-basic-offset: 4
*End:
*vim600: noet sw=4 ts=4 fdm=marker
*vim<600: noet sw=4 ts=4
*/

File Metadata

Mime Type
text/x-c
Expires
Sun, Nov 15, 4:25 AM (1 d, 17 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
132253
Default Alt Text
amqp_exchange.c (30 KB)

Event Timeline