jessequ
/
AMQP-CPP
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1 from CopernicaMarketingSoftware/master 

update 'master' branch from main repo
This commit is contained in:
Steven G 2018-01-31 15:19:04 +01:00 committed by GitHub
parent 9935f8414b 0cdb1fe500
commit fff1c65854
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
117 changed files with 823 additions and 654 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@ -13,3 +13,8 @@
*.la
*.a
*.a.*
/build
/.vscode
.atom-build.cson
.atom-dbg.cson
/bin

91
.travis.yml Normal file
View File

@ -0,0 +1,91 @@
################
# Config
################
# C++ project
language: cpp
dist: trusty
sudo: required
group: edge
################
# Services
################
services:
- docker
################
# Build matrix
################
matrix:
include:
################
# Linux / GCC
################
- os: linux
env:
- COMPILER_PACKAGE=g++-6
- C_COMPILER=gcc-6
- CXX_COMPILER=g++-6
- os: linux
compiler: gcc
env:
- COMPILER_PACKAGE=g++-7
- C_COMPILER=gcc-7
- CXX_COMPILER=g++-7
################
# Linux / Clang
################
- os: linux
env:
- COMPILER_PACKAGE=clang-4.0
- C_COMPILER=clang-4.0
- CXX_COMPILER=clang++-4.0
- os: linux
env:
- COMPILER_PACKAGE=clang-5.0
- C_COMPILER=clang-5.0
- CXX_COMPILER=clang++-5.0
before_install:
# Show OS/compiler version (this may not be the same OS as the Docker container)
- uname -a
# Use an artful container - gives us access to latest compilers.
- docker run -d --name ubuntu-test-container -v $(pwd):/travis ubuntu:artful tail -f /dev/null
- docker ps
install:
# Create our container
- docker exec -t ubuntu-test-container bash -c "apt-get update -y &&
apt-get --no-install-recommends install -y software-properties-common cmake
ninja-build libboost-all-dev libev-dev libuv1-dev ninja-build $COMPILER_PACKAGE &&
apt-get -y clean && rm -rf /var/lib/apt/lists/*"
################
# Build / Test
################
script:
# Run the container that we created and build the code
- docker exec -t ubuntu-test-container bash -c "cd /travis &&
export CC=/usr/bin/$C_COMPILER &&
export CXX=/usr/bin/$CXX_COMPILER &&
mkdir build.release && cd build.release &&
cmake .. ${CMAKE_OPTIONS} -DAMQP-CPP_BUILD_EXAMPLES=ON -DAMQP-CPP_LINUX_TCP=ON -GNinja &&
cmake --build . --config Release &&
cd .."

102
CMakeLists.txt
View File

@ -1,11 +1,36 @@
cmake_minimum_required(VERSION 2.8)
# Builds AMQP-CPP
#
# Options:
#
# - AMQP-CPP_BUILD_SHARED (default OFF)
# ON: Build shared lib
# OFF: Build static lib
#
# - AMQP-CPP_LINUX_TCP (default OFF)
# ON: Build posix handler implementation
# OFF: Don't build posix handler implementation
cmake_minimum_required(VERSION 3.1)
# project name
project(amqpcpp)
# ensure c++11 on all compilers
include(set_cxx_norm.cmake)
set_cxx_norm (${CXX_NORM_CXX11})
# build options
option(AMQP-CPP_BUILD_SHARED "Build shared library. If off, build will be static." OFF)
option(AMQP-CPP_LINUX_TCP "Build linux sockets implementation." OFF)
option(AMQP-CPP_BUILD_EXAMPLES "Build amqpcpp examples" OFF)
# ensure c++11 on all compilers
set (CMAKE_CXX_STANDARD 11)
# add source files
# ------------------------------------------------------------------------------------------------------
# set include/ as include directory
include_directories(${CMAKE_SOURCE_DIR}/include)
# macro that adds a list of provided source files to a list called SRCS.
# if variable SRCS does not yet exist, it is created.
macro (add_sources)
file (RELATIVE_PATH _relPath "${PROJECT_SOURCE_DIR}" "${CMAKE_CURRENT_SOURCE_DIR}")
foreach (_src ${ARGN})
@ -21,36 +46,59 @@ macro (add_sources)
endif()
endmacro()
# add source files
add_subdirectory(src)
add_subdirectory(include)
if(AMQP-CPP_LINUX_TCP)
add_subdirectory(src/linux_tcp)
endif()
option(BUILD_SHARED "build shared library" OFF)
# potentially build the examples
if(AMQP-CPP_BUILD_EXAMPLES)
add_subdirectory(examples)
endif()
if(BUILD_SHARED)
add_library(amqpcpp SHARED ${SRCS})
set_target_properties(amqpcpp PROPERTIES SOVERSION 2.7)
install(TARGETS amqpcpp
# settings for specific compilers
# ------------------------------------------------------------------------------------------------------
# we have to prevent windows from defining the max macro.
if (WIN32)
add_definitions(-DNOMINMAX)
endif()
# build targets
# ------------------------------------------------------------------------------------------------------
# set output directory
set(LIBRARY_OUTPUT_PATH ${CMAKE_SOURCE_DIR}/bin)
if(AMQP-CPP_BUILD_SHARED)
# create shared lib
add_library(${PROJECT_NAME} SHARED ${SRCS})
# set shared lib version
set_target_properties(${PROJECT_NAME} PROPERTIES SOVERSION 2.8)
else()
# create static lib
add_library(${PROJECT_NAME} STATIC ${SRCS})
endif()
# install rules
# ------------------------------------------------------------------------------------------------------
if(AMQP-CPP_BUILD_SHARED)
# copy shared lib and its static counter part
install(TARGETS ${PROJECT_NAME}
ARCHIVE DESTINATION lib
LIBRARY DESTINATION lib
RUNTIME DESTINATION lib
)
else()
add_library(amqpcpp STATIC ${SRCS})
install(TARGETS amqpcpp
# copy static lib
install(TARGETS ${PROJECT_NAME}
ARCHIVE DESTINATION lib
)
endif()
Include_directories(${PROJECT_SOURCE_DIR})
install(DIRECTORY include/ DESTINATION include/amqpcpp
# copy header files
install(DIRECTORY include/amqpcpp/ DESTINATION include/amqpcpp
FILES_MATCHING PATTERN "*.h")
install(FILES amqpcpp.h DESTINATION include)
option(BUILD_TUTORIALS "build rabbitmq tutorials" OFF)
if(BUILD_TUTORIALS)
# add_subdirectory(examples/rabbitmq_tutorials)
endif()
set(AMQP-CPP_INCLUDE_PATH ${CMAKE_CURRENT_SOURCE_DIR})
set(AMQP-CPP_INCLUDE_PATH ${CMAKE_CURRENT_SOURCE_DIR} PARENT_SCOPE)
if (WIN32)
add_definitions(-DNOMINMAX -DWIN32_LEAN_AND_MEAN)
endif()
install(FILES include/amqpcpp.h DESTINATION include)

6
Makefile
View File

@ -25,9 +25,11 @@ clean:
install:
mkdir -p ${INCLUDE_DIR}/$(LIBRARY_NAME)
mkdir -p ${INCLUDE_DIR}/$(LIBRARY_NAME)/linux_tcp
mkdir -p ${LIBRARY_DIR}
cp -f $(LIBRARY_NAME).h ${INCLUDE_DIR}
cp -f include/*.h ${INCLUDE_DIR}/$(LIBRARY_NAME)
cp -f include/$(LIBRARY_NAME).h ${INCLUDE_DIR}
cp -f include/amqpcpp/*.h ${INCLUDE_DIR}/$(LIBRARY_NAME)
cp -f include/amqpcpp/linux_tcp/*.h ${INCLUDE_DIR}/$(LIBRARY_NAME)/linux_tcp
-cp -f src/lib$(LIBRARY_NAME).so.$(VERSION) ${LIBRARY_DIR}
-cp -f src/lib$(LIBRARY_NAME).a.$(VERSION) ${LIBRARY_DIR}
ln -r -s -f $(LIBRARY_DIR)/lib$(LIBRARY_NAME).so.$(VERSION) $(LIBRARY_DIR)/lib$(LIBRARY_NAME).so.$(SONAME)

38
README.md
View File

@ -1,6 +1,8 @@
AMQP-CPP
========
[![Build Status](https://travis-ci.org/CopernicaMarketingSoftware/AMQP-CPP.svg?branch=master)](https://travis-ci.org/CopernicaMarketingSoftware/AMQP-CPP)
AMQP-CPP is a C++ library for communicating with a RabbitMQ message broker. The
library can be used to parse incoming data from a RabbitMQ server, and to
generate frames that can be sent to a RabbitMQ server.
@ -62,20 +64,41 @@ Then check out our other commercial and open source solutions:
INSTALLING
==========
AMQP-CPP comes with an optional Linux-only TCP module that takes care of the network part required for the AMQP-CPP core library. If you use this module, you are required to link with `pthread`.
If you are on some kind of Linux environment, installing the library is as easy
There are two methods to compile AMQP-CPP: CMake and Make. CMake is platform portable, but the Makefile only works on Linux. Building of a shared library is currently not supported on Windows.
After building there are two relevant files to include when using the library.
File|Include when?
----|------------
amqpcpp.h|Always
amqpcpp/linux_tcp.h|If using the Linux-only TCP module
On Windows you are required to define `NOMINMAX` when compiling code that includes public AMQP-CPP header files.
## CMake
The CMake file supports both building and installing. You can choose not to use the install functionality, and instead manually use the build output at `bin/`. Keep in mind that the TCP module is only supported for Linux. An example install method would be:
``` bash
mkdir build
cd build
cmake .. [-DBUILD_SHARED] [-DLINUX_TCP]
cmake --build .. --target install
```
Option|Default|Meaning
------|-------|-------
BUILD_SHARED|OFF|Static lib(ON) or shared lib(OFF)? Shared is not supported on Windows.
LINUX_TCP|OFF|Should the Linux-only TCP module be built?
## Make
Installing the library is as easy
as running `make` and `make install`. This will install the full version of
the AMQP-CPP, including the system specific TCP module. If you do not need the
additional TCP module (because you take care of handling the network stuff
yourself), you can also compile a pure form of the library. Use `make pure`
and `make install` for that.
For users on a non-Linux environment: this library is known to work on your
environment too (after all, it does not do any operating specific system calls),
but it might take some extra effort to get your compiler to compile it. Please
send in your pull requests once you have it running, so that others can benefit
from your experiences.
When you compile an application that uses the AMQP-CPP library, do not
forget to link with the library. For gcc and clang the linker flag is -lamqpcpp.
If you use the fullblown version of AMQP-CPP (with the TCP module), you also
@ -992,4 +1015,3 @@ class that can be directly plugged into libev, libevent and libuv event loops.
For performance reasons, we need to investigate if we can limit the number of times
an incoming or outgoing messages is copied.

77
amqpcpp.h
View File

@ -1,77 +0,0 @@
/**
* AMQP.h
*
* Starting point for all includes of the Copernica AMQP library
*
* @documentation public
*/
#pragma once
// base C++ include files
#include <vector>
#include <string>
#include <memory>
#include <map>
#include <unordered_map>
#include <queue>
#include <limits>
#include <cstddef>
#include <cstring>
#include <stdexcept>
#include <utility>
#include <iostream>
#include <algorithm>
#include <functional>
// base C include files
#include <stdint.h>
#include <math.h>
// forward declarations
#include <amqpcpp/classes.h>
// utility classes
#include <amqpcpp/endian.h>
#include <amqpcpp/buffer.h>
#include <amqpcpp/bytebuffer.h>
#include <amqpcpp/receivedframe.h>
#include <amqpcpp/outbuffer.h>
#include <amqpcpp/watchable.h>
#include <amqpcpp/monitor.h>
// amqp types
#include <amqpcpp/field.h>
#include <amqpcpp/numericfield.h>
#include <amqpcpp/decimalfield.h>
#include <amqpcpp/stringfield.h>
#include <amqpcpp/booleanset.h>
#include <amqpcpp/fieldproxy.h>
#include <amqpcpp/table.h>
#include <amqpcpp/array.h>
// envelope for publishing and consuming
#include <amqpcpp/metadata.h>
#include <amqpcpp/envelope.h>
#include <amqpcpp/message.h>
// mid level includes
#include <amqpcpp/exchangetype.h>
#include <amqpcpp/flags.h>
#include <amqpcpp/callbacks.h>
#include <amqpcpp/deferred.h>
#include <amqpcpp/deferredconsumer.h>
#include <amqpcpp/deferredqueue.h>
#include <amqpcpp/deferreddelete.h>
#include <amqpcpp/deferredcancel.h>
#include <amqpcpp/deferredget.h>
#include <amqpcpp/channelimpl.h>
#include <amqpcpp/channel.h>
#include <amqpcpp/login.h>
#include <amqpcpp/address.h>
#include <amqpcpp/connectionhandler.h>
#include <amqpcpp/connectionimpl.h>
#include <amqpcpp/connection.h>
#include <amqpcpp/tcphandler.h>
#include <amqpcpp/tcpconnection.h>
#include <amqpcpp/tcpchannel.h>

19
appveyor.yml Normal file
View File

@ -0,0 +1,19 @@
version: '1.0.{build}'
image: Visual Studio 2017
platform:
- x64
configuration:
- Release
- Debug
install:
- git submodule update --init --recursive
before_build:
- cmake -G "Visual Studio 15 2017 Win64" .
build:
project: $(APPVEYOR_BUILD_FOLDER)\amqpcpp.sln

32
examples/CMakeLists.txt Normal file
View File

@ -0,0 +1,32 @@
###################################
# Boost
###################################
add_executable(amqpcpp_boost_example libboostasio.cpp)
add_dependencies(amqpcpp_boost_example amqpcpp)
target_link_libraries(amqpcpp_boost_example amqpcpp boost_system pthread)
###################################
# Libev
###################################
add_executable(amqpcpp_libev_example libev.cpp)
add_dependencies(amqpcpp_libev_example amqpcpp)
target_link_libraries(amqpcpp_libev_example amqpcpp ev pthread)
###################################
# Libuv
###################################
add_executable(amqpcpp_libuv_example libuv.cpp)
add_dependencies(amqpcpp_libuv_example amqpcpp)
target_link_libraries(amqpcpp_libuv_example amqpcpp uv pthread)

0
tests/libboostasio.cpp → examples/libboostasio.cpp
View File

0
tests/libev.cpp → examples/libev.cpp
View File

0
tests/libevent.cpp → examples/libevent.cpp
View File

0
tests/libuv.cpp → examples/libuv.cpp
View File

54
include/CMakeLists.txt
View File

@ -1,54 +0,0 @@
add_sources(
address.h
addresses.h
array.h
booleanset.h
buffer.h
bytebuffer.h
callbacks.h
channel.h
channelimpl.h
classes.h
connection.h
connectionhandler.h
connectionimpl.h
copiedbuffer.h
decimalfield.h
deferred.h
deferredcancel.h
deferredconsumer.h
deferredconsumerbase.h
deferreddelete.h
deferredget.h
deferredqueue.h
endian.h
entityimpl.h
envelope.h
exception.h
exchangetype.h
field.h
fieldproxy.h
flags.h
frame.h
libboostasio.h
libev.h
libevent.h
libuv.h
login.h
message.h
metadata.h
monitor.h
numericfield.h
outbuffer.h
protocolexception.h
receivedframe.h
stack_ptr.h
stringfield.h
table.h
tcpchannel.h
tcpconnection.h
tcpdefines.h
tcphandler.h
watchable.h
)

84
include/amqpcpp.h Normal file
View File

@ -0,0 +1,84 @@
/**
* AMQP.h
*
* Starting point for all includes of the Copernica AMQP library
*
* @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com>
* @copyright 2015 - 2018 Copernica BV
*/
#pragma once
// base C++ include files
#include <vector>
#include <string>
#include <memory>
#include <map>
#include <unordered_map>
#include <queue>
#include <limits>
#include <cstddef>
#include <cstring>
#include <stdexcept>
#include <utility>
#include <iostream>
#include <algorithm>
#include <functional>
// base C include files
#include <stdint.h>
#include <math.h>
// fix strcasecmp on non linux platforms
#if (defined(_WIN16) || defined(_WIN32) || defined(_WIN64) || defined(__WINDOWS__)) && !defined(__CYGWIN__)
#define strcasecmp _stricmp
#endif
// forward declarations
#include "amqpcpp/classes.h"
// utility classes
#include "amqpcpp/endian.h"
#include "amqpcpp/buffer.h"
#include "amqpcpp/bytebuffer.h"
#include "amqpcpp/receivedframe.h"
#include "amqpcpp/outbuffer.h"
#include "amqpcpp/watchable.h"
#include "amqpcpp/monitor.h"
// amqp types
#include "amqpcpp/field.h"
#include "amqpcpp/numericfield.h"
#include "amqpcpp/decimalfield.h"
#include "amqpcpp/stringfield.h"
#include "amqpcpp/booleanset.h"
#include "amqpcpp/fieldproxy.h"
#include "amqpcpp/table.h"
#include "amqpcpp/array.h"
// envelope for publishing and consuming
#include "amqpcpp/metadata.h"
#include "amqpcpp/envelope.h"
#include "amqpcpp/message.h"
// mid level includes
#include "amqpcpp/exchangetype.h"
#include "amqpcpp/flags.h"
#include "amqpcpp/callbacks.h"
#include "amqpcpp/deferred.h"
#include "amqpcpp/deferredconsumer.h"
#include "amqpcpp/deferredqueue.h"
#include "amqpcpp/deferreddelete.h"
#include "amqpcpp/deferredcancel.h"
#include "amqpcpp/deferredget.h"
#include "amqpcpp/channelimpl.h"
#include "amqpcpp/channel.h"
#include "amqpcpp/login.h"
#include "amqpcpp/address.h"
#include "amqpcpp/connectionhandler.h"
#include "amqpcpp/connectionimpl.h"
#include "amqpcpp/connection.h"
// tcp level includes
#include "amqpcpp/linux_tcp.h"

0
include/address.h → include/amqpcpp/address.h
View File

0
include/addresses.h → include/amqpcpp/addresses.h
View File

0
include/array.h → include/amqpcpp/array.h
View File

0
include/booleanset.h → include/amqpcpp/booleanset.h
View File

0
include/buffer.h → include/amqpcpp/buffer.h
View File

0
include/bytebuffer.h → include/amqpcpp/bytebuffer.h
View File

0
include/callbacks.h → include/amqpcpp/callbacks.h
View File

0
include/channel.h → include/amqpcpp/channel.h
View File

0
include/channelimpl.h → include/amqpcpp/channelimpl.h
View File

0
include/classes.h → include/amqpcpp/classes.h
View File

0
include/connection.h → include/amqpcpp/connection.h
View File

0
include/connectionhandler.h → include/amqpcpp/connectionhandler.h
View File

0
include/connectionimpl.h → include/amqpcpp/connectionimpl.h
View File

0
include/copiedbuffer.h → include/amqpcpp/copiedbuffer.h
View File

0
include/decimalfield.h → include/amqpcpp/decimalfield.h
View File

0
include/deferred.h → include/amqpcpp/deferred.h
View File

0
include/deferredcancel.h → include/amqpcpp/deferredcancel.h
View File

0
include/deferredconsumer.h → include/amqpcpp/deferredconsumer.h
View File

0
include/deferredconsumerbase.h → include/amqpcpp/deferredconsumerbase.h
View File

0
include/deferreddelete.h → include/amqpcpp/deferreddelete.h
View File

0
include/deferredget.h → include/amqpcpp/deferredget.h
View File

0
include/deferredqueue.h → include/amqpcpp/deferredqueue.h
View File

2
include/endian.h → include/amqpcpp/endian.h
View File

@ -72,7 +72,7 @@
#define be32toh(x) ntohl(x)
#define le32toh(x) (x)
#define htobe64(x) htonll(x)
#define htobe64(x) ((1==htonl(1)) ? (x) : ((uint64_t)htonl((x) & 0xFFFFFFFF) << 32) | htonl((x) >> 32))
#define htole64(x) (x)
#define be64toh(x) ntohll(x)
#define le64toh(x) (x)

0
include/entityimpl.h → include/amqpcpp/entityimpl.h
View File

0
include/envelope.h → include/amqpcpp/envelope.h
View File

0
include/exception.h → include/amqpcpp/exception.h
View File

0
include/exchangetype.h → include/amqpcpp/exchangetype.h
View File

0
include/field.h → include/amqpcpp/field.h
View File

0
include/fieldproxy.h → include/amqpcpp/fieldproxy.h
View File

0
include/flags.h → include/amqpcpp/flags.h
View File

0
include/frame.h → include/amqpcpp/frame.h
View File

252
include/libboostasio.h → include/amqpcpp/libboostasio.h
View File

@ -25,36 +25,16 @@
#include <boost/asio/deadline_timer.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include "amqpcpp/linux_tcp.h"
///////////////////////////////////////////////////////////////////
#define STRAND_SOCKET_HANDLER(_fn) \
[fn = _fn, strand = _strand](const boost::system::error_code &ec, \
const std::size_t bytes_transferred) \
{ \
const std::shared_ptr<boost::asio::io_service::strand> apStrand = strand.lock(); \
if (!apStrand) \
{ \
fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled),std::size_t{0}); \
return; \
} \
\
apStrand->dispatch(boost::bind(fn,ec,bytes_transferred)); \
}
///////////////////////////////////////////////////////////////////
#define STRAND_TIMER_HANDLER(_fn) \
[fn = _fn, strand = _strand](const boost::system::error_code &ec) \
{ \
const std::shared_ptr<boost::asio::io_service::strand> apStrand = strand.lock(); \
if (!apStrand) \
{ \
fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled)); \
return; \
} \
\
apStrand->dispatch(boost::bind(fn,ec)); \
}
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
#define PTR_FROM_THIS weak_from_this
#else
#define PTR_FROM_THIS shared_from_this
#endif
/**
* Set up namespace
@ -82,11 +62,13 @@ private:
*/
boost::asio::io_service & _ioservice;
typedef std::weak_ptr<boost::asio::io_service::strand> strand_weak_ptr;
/**
* The boost asio io_service::strand managed pointer.
* @var class std::shared_ptr<boost::asio::io_service>
*/
std::weak_ptr<boost::asio::io_service::strand> _strand;
strand_weak_ptr _wpstrand;
/**
* The boost tcp socket.
@ -120,6 +102,66 @@ private:
*/
bool _write_pending{false};
using handler_cb = boost::function<void(boost::system::error_code,std::size_t)>;
using io_handler = boost::function<void(const boost::system::error_code&, const std::size_t)>;
/**
* Builds a io handler callback that executes the io callback in a strand.
* @param io_handler The handler callback to dispatch
* @return handler_cb A function wrapping the execution of the handler function in a io_service::strand.
*/
handler_cb get_dispatch_wrapper(io_handler fn)
{
const strand_weak_ptr wpstrand = _wpstrand;
return [fn, wpstrand](const boost::system::error_code &ec, const std::size_t bytes_transferred)
{
const strand_shared_ptr strand = wpstrand.lock();
if (!strand)
{
fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled), std::size_t{0});
return;
}
strand->dispatch(boost::bind(fn, ec, bytes_transferred));
};
}
/**
* Binds and returns a read handler for the io operation.
* @param connection The connection being watched.
* @param fd The file descripter being watched.
* @return handler callback
*/
handler_cb get_read_handler(TcpConnection *const connection, const int fd)
{
auto fn = boost::bind(&Watcher::read_handler,
this,
_1,
_2,
PTR_FROM_THIS(),
connection,
fd);
return get_dispatch_wrapper(fn);
}
/**
* Binds and returns a read handler for the io operation.
* @param connection The connection being watched.
* @param fd The file descripter being watched.
* @return handler callback
*/
handler_cb get_write_handler(TcpConnection *const connection, const int fd)
{
auto fn = boost::bind(&Watcher::write_handler,
this,
_1,
_2,
PTR_FROM_THIS(),
connection,
fd);
return get_dispatch_wrapper(fn);
}
/**
* Handler method that is called by boost's io_service when the socket pumps a read event.
* @param ec The status of the callback.
@ -148,20 +190,9 @@ private:
_read_pending = true;
_socket.async_read_some(boost::asio::null_buffers(),
STRAND_SOCKET_HANDLER(
boost::bind(&Watcher::read_handler,
this,
boost::arg<1>(),
boost::arg<2>(),
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
weak_from_this(),
#else
shared_from_this(),
#endif
connection,
fd)));
_socket.async_read_some(
boost::asio::null_buffers(),
get_read_handler(connection, fd));
}
}
@ -193,20 +224,9 @@ private:
_write_pending = true;
_socket.async_write_some(boost::asio::null_buffers(),
STRAND_SOCKET_HANDLER(
boost::bind(&Watcher::write_handler,
this,
boost::arg<1>(),
boost::arg<2>(),
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
weak_from_this(),
#else
shared_from_this(),
#endif
connection,
fd)));
_socket.async_write_some(
boost::asio::null_buffers(),
get_write_handler(connection, fd));
}
}
@ -215,14 +235,14 @@ private:
* Constructor- initialises the watcher and assigns the filedescriptor to
* a boost socket for monitoring.
* @param io_service The boost io_service
* @param strand A weak pointer to a io_service::strand instance.
* @param wpstrand A weak pointer to a io_service::strand instance.
* @param fd The filedescriptor being watched
*/
Watcher(boost::asio::io_service &io_service,
const std::weak_ptr<boost::asio::io_service::strand> strand,
const strand_weak_ptr wpstrand,
const int fd) :
_ioservice(io_service),
_strand(strand),
_wpstrand(wpstrand),
_socket(_ioservice)
{
_socket.assign(fd);
@ -262,20 +282,9 @@ private:
{
_read_pending = true;
_socket.async_read_some(boost::asio::null_buffers(),
STRAND_SOCKET_HANDLER(
boost::bind(&Watcher::read_handler,
this,
boost::arg<1>(),
boost::arg<2>(),
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
weak_from_this(),
#else
shared_from_this(),
#endif
connection,
fd)));
_socket.async_read_some(
boost::asio::null_buffers(),
get_read_handler(connection, fd));
}
// 2. Handle writes?
@ -286,20 +295,9 @@ private:
{
_write_pending = true;
_socket.async_write_some(boost::asio::null_buffers(),
STRAND_SOCKET_HANDLER(
boost::bind(&Watcher::write_handler,
this,
boost::arg<1>(),
boost::arg<2>(),
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
weak_from_this(),
#else
shared_from_this(),
#endif
connection,
fd)));
_socket.async_write_some(
boost::asio::null_buffers(),
get_write_handler(connection, fd));
}
}
};
@ -317,11 +315,13 @@ private:
*/
boost::asio::io_service & _ioservice;
typedef std::weak_ptr<boost::asio::io_service::strand> strand_weak_ptr;
/**
* The boost asio io_service::strand managed pointer.
* @var class std::shared_ptr<boost::asio::io_service>
*/
std::weak_ptr<boost::asio::io_service::strand> _strand;
strand_weak_ptr _wpstrand;
/**
* The boost asynchronous deadline timer.
@ -329,11 +329,42 @@ private:
*/
boost::asio::deadline_timer _timer;
using handler_fn = boost::function<void(boost::system::error_code)>;
/**
* Binds and returns a lamba function handler for the io operation.
* @param connection The connection being watched.
* @param timeout The file descripter being watched.
* @return handler callback
*/
handler_fn get_handler(TcpConnection *const connection, const uint16_t timeout)
{
const auto fn = boost::bind(&Timer::timeout,
this,
_1,
PTR_FROM_THIS(),
connection,
timeout);
const strand_weak_ptr wpstrand = _wpstrand;
return [fn, wpstrand](const boost::system::error_code &ec)
{
const strand_shared_ptr strand = wpstrand.lock();
if (!strand)
{
fn(boost::system::errc::make_error_code(boost::system::errc::operation_canceled));
return;
}
strand->dispatch(boost::bind(fn, ec));
};
}
/**
* Callback method that is called by libev when the timer expires
* @param ec error code returned from loop
* @param loop The loop in which the event was triggered
* @param timer Internal timer object
* @param revents The events that triggered this call
* @param connection
* @param timeout
*/
void timeout(const boost::system::error_code &ec,
std::weak_ptr<Timer> awpThis,
@ -357,18 +388,7 @@ private:
_timer.expires_at(_timer.expires_at() + boost::posix_time::seconds(timeout));
// Posts the timer event
_timer.async_wait(STRAND_TIMER_HANDLER(
boost::bind(&Timer::timeout,
this,
boost::arg<1>(),
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
weak_from_this(),
#else
shared_from_this(),
#endif
connection,
timeout)));
_timer.async_wait(get_handler(connection, timeout));
}
}
@ -385,12 +405,12 @@ private:
/**
* Constructor
* @param io_service The boost asio io_service.
* @param strand A weak pointer to a io_service::strand instance.
* @param wpstrand A weak pointer to a io_service::strand instance.
*/
Timer(boost::asio::io_service &io_service,
const std::weak_ptr<boost::asio::io_service::strand> strand) :
const strand_weak_ptr wpstrand) :
_ioservice(io_service),
_strand(strand),
_wpstrand(wpstrand),
_timer(_ioservice)
{
@ -423,19 +443,11 @@ private:
// stop timer in case it was already set
stop();
// Reschedule the timer for the future:
_timer.expires_from_now(boost::posix_time::seconds(timeout));
_timer.async_wait(STRAND_TIMER_HANDLER(
boost::bind(&Timer::timeout,
this,
boost::arg<1>(),
// C++17 has 'weak_from_this()' support.
#if __cplusplus >= 201701L
weak_from_this(),
#else
shared_from_this(),
#endif
connection,
timeout)));
// Posts the timer event
_timer.async_wait(get_handler(connection, timeout));
}
};
@ -445,11 +457,13 @@ private:
*/
boost::asio::io_service & _ioservice;
typedef std::shared_ptr<boost::asio::io_service::strand> strand_shared_ptr;
/**
* The boost asio io_service::strand managed pointer.
* @var class std::shared_ptr<boost::asio::io_service>
*/
std::shared_ptr<boost::asio::io_service::strand> _strand;
strand_shared_ptr _strand;
/**

4
include/libev.h → include/amqpcpp/libev.h
View File

@ -8,7 +8,7 @@
* Compile with: "g++ -std=c++11 libev.cpp -lamqpcpp -lev -lpthread"
*
* @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com>
* @copyright 2015 Copernica BV
* @copyright 2015 - 2018 Copernica BV
*/
/**
@ -21,6 +21,8 @@
*/
#include <ev.h>
#include "amqpcpp/linux_tcp.h"
/**
* Set up namespace
*/

0
include/libevent.h → include/amqpcpp/libevent.h
View File

2
include/libuv.h → include/amqpcpp/libuv.h
View File

@ -21,6 +21,8 @@
*/
#include <uv.h>
#include "amqpcpp/linux_tcp.h"
/**
* Set up namespace
*/

3
include/amqpcpp/linux_tcp.h Normal file
View File

@ -0,0 +1,3 @@
#include "linux_tcp/tcphandler.h"
#include "linux_tcp/tcpconnection.h"
#include "linux_tcp/tcpchannel.h"

0
include/tcpchannel.h → include/amqpcpp/linux_tcp/tcpchannel.h
View File

0
include/tcpconnection.h → include/amqpcpp/linux_tcp/tcpconnection.h
View File

0
include/tcpdefines.h → include/amqpcpp/linux_tcp/tcpdefines.h
View File

0
include/tcphandler.h → include/amqpcpp/linux_tcp/tcphandler.h
View File

0
include/login.h → include/amqpcpp/login.h
View File

11
include/message.h → include/amqpcpp/message.h
View File

@ -21,6 +21,7 @@
#include "envelope.h"
#include <limits>
#include <stdexcept>
#include <algorithm>
/**
* Set up namespace
@ -93,10 +94,10 @@ protected:
size = std::min(size, _bodySize - _filled);
// append more data
memcpy(_body + _filled, buffer, size);
memcpy(_body + _filled, buffer, (size_t)size);
// update filled data
_filled += size;
_filled += (size_t)size;
}
else if (size >= _bodySize)
{
@ -107,16 +108,16 @@ protected:
else
{
// allocate the buffer
_body = (char *)malloc(_bodySize);
_body = (char *)malloc((size_t)_bodySize);
// remember that the buffer was allocated, so that the destructor can get rid of it
_allocated = true;
// append more data
memcpy(_body, buffer, std::min(size, _bodySize));
memcpy(_body, buffer, std::min((size_t)size, (size_t)_bodySize));
// update filled data
_filled = std::min(size, _bodySize);
_filled = std::min((size_t)size, (size_t)_bodySize);
}
// check if we're done

28
include/metadata.h → include/amqpcpp/metadata.h
View File

@ -302,20 +302,20 @@ public:
// the result (2 for the two boolean sets)
uint32_t result = 2;
if (hasExpiration()) result += _expiration.size();
if (hasReplyTo()) result += _replyTo.size();
if (hasCorrelationID()) result += _correlationID.size();
if (hasPriority()) result += _priority.size();
if (hasDeliveryMode()) result += _deliveryMode.size();
if (hasHeaders()) result += _headers.size();
if (hasContentEncoding()) result += _contentEncoding.size();
if (hasContentType()) result += _contentType.size();
if (hasClusterID()) result += _clusterID.size();
if (hasAppID()) result += _appID.size();
if (hasUserID()) result += _userID.size();
if (hasTypeName()) result += _typeName.size();
if (hasTimestamp()) result += _timestamp.size();
if (hasMessageID()) result += _messageID.size();
if (hasExpiration()) result += (uint32_t)_expiration.size();
if (hasReplyTo()) result += (uint32_t)_replyTo.size();
if (hasCorrelationID()) result += (uint32_t)_correlationID.size();
if (hasPriority()) result += (uint32_t)_priority.size();
if (hasDeliveryMode()) result += (uint32_t)_deliveryMode.size();
if (hasHeaders()) result += (uint32_t)_headers.size();
if (hasContentEncoding()) result += (uint32_t)_contentEncoding.size();
if (hasContentType()) result += (uint32_t)_contentType.size();
if (hasClusterID()) result += (uint32_t)_clusterID.size();
if (hasAppID()) result += (uint32_t)_appID.size();
if (hasUserID()) result += (uint32_t)_userID.size();
if (hasTypeName()) result += (uint32_t)_typeName.size();
if (hasTimestamp()) result += (uint32_t)_timestamp.size();
if (hasMessageID()) result += (uint32_t)_messageID.size();
// done
return result;

0
include/monitor.h → include/amqpcpp/monitor.h
View File

19
include/numericfield.h → include/amqpcpp/numericfield.h
View File

@ -42,6 +42,8 @@ private:
T _value;
public:
using Type = T;
/**
* Default constructor, assign 0
*/
@ -116,14 +118,14 @@ public:
* Get the value
* @return mixed
*/
operator uint8_t () const override { return _value; }
operator uint16_t() const override { return _value; }
operator uint32_t() const override { return _value; }
operator uint64_t() const override { return _value; }
operator int8_t () const override { return _value; }
operator int16_t () const override { return _value; }
operator int32_t () const override { return _value; }
operator int64_t () const override { return _value; }
operator uint8_t () const override { return (uint8_t)_value; }
operator uint16_t() const override { return (uint16_t)_value; }
operator uint32_t() const override { return (uint32_t)_value; }
operator uint64_t() const override { return (uint64_t)_value; }
operator int8_t () const override { return (int8_t)_value; }
operator int16_t () const override { return (int16_t)_value; }
operator int32_t () const override { return (int32_t)_value; }
operator int64_t () const override { return (int64_t)_value; }
/**
* Get the value
@ -213,4 +215,3 @@ typedef NumericField<double, 'd'> Double;
* end namespace
*/
}

0
include/outbuffer.h → include/amqpcpp/outbuffer.h
View File

0
include/protocolexception.h → include/amqpcpp/protocolexception.h
View File

0
include/receivedframe.h → include/amqpcpp/receivedframe.h
View File

0
include/stack_ptr.h → include/amqpcpp/stack_ptr.h
View File

5
include/stringfield.h → include/amqpcpp/stringfield.h
View File

@ -119,7 +119,7 @@ public:
virtual size_t size() const override
{
// find out size of the size parameter
T size(_data.size());
T size((typename T::Type)_data.size());
// size of the uint8 or uint32 + the actual string size
return size.size() + _data.size();
@ -160,7 +160,7 @@ public:
virtual void fill(OutBuffer& buffer) const override
{
// create size
T size(_data.size());
T size((typename T::Type)_data.size());
// first, write down the size of the string
size.fill(buffer);
@ -210,4 +210,3 @@ typedef StringField<ULong, 'S'> LongString;
* end namespace
*/
}

0
include/table.h → include/amqpcpp/table.h
View File

0
include/watchable.h → include/amqpcpp/watchable.h
View File

55
set_cxx_norm.cmake
View File

@ -1,55 +0,0 @@
# Version
cmake_minimum_required(VERSION 2.6.3)
set(CXX_NORM_CXX98 1) # C++98
set(CXX_NORM_CXX03 2) # C++03
set(CXX_NORM_CXX11 3) # C++11
# - Set the wanted C++ norm
# Adds the good argument to the command line in function of the compiler
# Currently only works with g++ and clang++
macro(set_cxx_norm NORM)
# Extract c++ compiler --version output
exec_program(
${CMAKE_CXX_COMPILER}
ARGS --version
OUTPUT_VARIABLE _compiler_output
)
# Keep only the first line
string(REGEX REPLACE
"(\n.*$)"
""
cxx_compiler_version "${_compiler_output}"
)
# Extract the version number
string(REGEX REPLACE
"([^0-9.])|([0-9.][^0-9.])"
""
cxx_compiler_version "${cxx_compiler_version}"
)
if(CMAKE_COMPILER_IS_GNUCXX)
if(${NORM} EQUAL ${CXX_NORM_CXX98})
add_definitions("-std=c++98")
elseif(${NORM} EQUAL ${CXX_NORM_CXX03})
add_definitions("-std=c++03")
elseif(${NORM} EQUAL ${CXX_NORM_CXX11})
if(${cxx_compiler_version} VERSION_LESS "4.7.0")
add_definitions("-std=c++0x")
else()
add_definitions("-std=c++11")
endif()
endif()
elseif(${CMAKE_CXX_COMPILER_ID} STREQUAL "Clang")
if(${NORM} EQUAL ${CXX_NORM_CXX11})
add_definitions("-std=c++11")
endif()
endif()
endmacro()

187
src/CMakeLists.txt
View File

@ -1,100 +1,91 @@
add_sources(
addressinfo.h
array.cpp
basicackframe.h
basiccancelframe.h
basiccancelokframe.h
basicconsumeframe.h
basicconsumeokframe.h
basicdeliverframe.h
basicframe.h
basicgetemptyframe.h
basicgetframe.h
basicgetokframe.h
basicheaderframe.h
basicnackframe.h
basicpublishframe.h
basicqosframe.h
basicqosokframe.h
basicrecoverasyncframe.h
basicrecoverframe.h
basicrecoverokframe.h
basicrejectframe.h
basicreturnframe.h
bodyframe.h
channelcloseframe.h
channelcloseokframe.h
channelflowframe.h
channelflowokframe.h
channelframe.h
channelimpl.cpp
channelopenframe.h
channelopenokframe.h
connectioncloseframe.h
connectioncloseokframe.h
connectionframe.h
connectionimpl.cpp
connectionopenframe.h
connectionopenokframe.h
connectionsecureframe.h
connectionsecureokframe.h
connectionstartframe.h
connectionstartokframe.h
connectiontuneframe.h
connectiontuneokframe.h
consumedmessage.h
deferredcancel.cpp
deferredconsumer.cpp
deferredconsumerbase.cpp
deferredget.cpp
exchangebindframe.h
exchangebindokframe.h
exchangedeclareframe.h
exchangedeclareokframe.h
exchangedeleteframe.h
exchangedeleteokframe.h
exchangeframe.h
exchangeunbindframe.h
exchangeunbindokframe.h
extframe.h
field.cpp
flags.cpp
framecheck.h
headerframe.h
heartbeatframe.h
includes.h
methodframe.h
passthroughbuffer.h
pipe.h
protocolheaderframe.h
queuebindframe.h
queuebindokframe.h
queuedeclareframe.h
queuedeclareokframe.h
queuedeleteframe.h
queuedeleteokframe.h
queueframe.h
queuepurgeframe.h
queuepurgeokframe.h
queueunbindframe.h
queueunbindokframe.h
receivedframe.cpp
reducedbuffer.h
returnedmessage.h
table.cpp
tcpclosed.h
tcpconnected.h
tcpconnection.cpp
tcpinbuffer.h
tcpoutbuffer.h
tcpresolver.h
tcpstate.h
transactioncommitframe.h
transactioncommitokframe.h
transactionframe.h
transactionrollbackframe.h
transactionrollbackokframe.h
transactionselectframe.h
transactionselectokframe.h
watchable.cpp
array.cpp
basicackframe.h
basiccancelframe.h
basiccancelokframe.h
basicconsumeframe.h
basicconsumeokframe.h
basicdeliverframe.h
basicframe.h
basicgetemptyframe.h
basicgetframe.h
basicgetokframe.h
basicheaderframe.h
basicnackframe.h
basicpublishframe.h
basicqosframe.h
basicqosokframe.h
basicrecoverasyncframe.h
basicrecoverframe.h
basicrecoverokframe.h
basicrejectframe.h
basicreturnframe.h
bodyframe.h
channelcloseframe.h
channelcloseokframe.h
channelflowframe.h
channelflowokframe.h
channelframe.h
channelimpl.cpp
channelopenframe.h
channelopenokframe.h
connectioncloseframe.h
connectioncloseokframe.h
connectionframe.h
connectionimpl.cpp
connectionopenframe.h
connectionopenokframe.h
connectionsecureframe.h
connectionsecureokframe.h
connectionstartframe.h
connectionstartokframe.h
connectiontuneframe.h
connectiontuneokframe.h
consumedmessage.h
deferredcancel.cpp
deferredconsumer.cpp
deferredconsumerbase.cpp
deferredget.cpp
exchangebindframe.h
exchangebindokframe.h
exchangedeclareframe.h
exchangedeclareokframe.h
exchangedeleteframe.h
exchangedeleteokframe.h
exchangeframe.h
exchangeunbindframe.h
exchangeunbindokframe.h
extframe.h
field.cpp
flags.cpp
framecheck.h
headerframe.h
heartbeatframe.h
includes.h
methodframe.h
passthroughbuffer.h
protocolheaderframe.h
queuebindframe.h
queuebindokframe.h
queuedeclareframe.h
queuedeclareokframe.h
queuedeleteframe.h
queuedeleteokframe.h
queueframe.h
queuepurgeframe.h
queuepurgeokframe.h
queueunbindframe.h
queueunbindokframe.h
receivedframe.cpp
reducedbuffer.h
returnedmessage.h
table.cpp
transactioncommitframe.h
transactioncommitokframe.h
transactionframe.h
transactionrollbackframe.h
transactionrollbackokframe.h
transactionselectframe.h
transactionselectokframe.h
watchable.cpp
)

5
src/Makefile
View File

@ -1,11 +1,11 @@
CPP = g++
RM = rm -f
CPPFLAGS = -Wall -c -I. -std=c++11 -MD
CPPFLAGS = -Wall -c -I../include -std=c++11 -MD
LD = g++
LD_FLAGS = -Wall -shared
SHARED_LIB = lib$(LIBRARY_NAME).so.$(VERSION)
STATIC_LIB = lib$(LIBRARY_NAME).a.$(VERSION)
SOURCES = $(wildcard *.cpp)
SOURCES = $(wildcard *.cpp) $(wildcard linux_tcp/*.cpp)
SHARED_OBJECTS = $(SOURCES:%.cpp=%.o)
STATIC_OBJECTS = $(SOURCES:%.cpp=%.s.o)
DEPENDENCIES = $(SOURCES:%.cpp=%.d)
@ -53,4 +53,3 @@ ${SHARED_OBJECTS}:
${STATIC_OBJECTS}:
${CPP} ${CPPFLAGS} -o $@ ${@:%.s.o=%.cpp}

4
src/array.cpp
View File

@ -29,7 +29,7 @@ Array::Array(ReceivedFrame &frame)
if (!field) continue;
// less bytes to read
charsToRead -= field->size();
charsToRead -= (uint32_t)field->size();
// add the additional field
_fields.push_back(std::shared_ptr<Field>(field));
@ -76,7 +76,7 @@ const Field &Array::get(uint8_t index) const
*/
uint32_t Array::count() const
{
return _fields.size();
return (uint32_t)_fields.size();
}
/**

2
src/basiccancelframe.h
View File

@ -59,7 +59,7 @@ public:
* @param noWait whether to wait for a response.
*/
BasicCancelFrame(uint16_t channel, const std::string& consumerTag, bool noWait = false) :
BasicFrame(channel, consumerTag.size() + 2), // 1 for extra string size, 1 for bool
BasicFrame(channel, (uint32_t)(consumerTag.size() + 2)), // 1 for extra string size, 1 for bool
_consumerTag(consumerTag),
_noWait(noWait) {}

2
src/basiccancelokframe.h
View File

@ -53,7 +53,7 @@ public:
* @param consumerTag holds the consumertag specified by client or server
*/
BasicCancelOKFrame(uint16_t channel, std::string& consumerTag) :
BasicFrame(channel, consumerTag.length() + 1), // add 1 byte for encoding the size of consumer tag
BasicFrame(channel, (uint32_t)(consumerTag.length() + 1)), // add 1 byte for encoding the size of consumer tag
_consumerTag(consumerTag)
{}

2
src/basicconsumeframe.h
View File

@ -86,7 +86,7 @@ public:
* @param filter additional arguments
*/
BasicConsumeFrame(uint16_t channel, const std::string& queueName, const std::string& consumerTag, bool noLocal = false, bool noAck = false, bool exclusive = false, bool noWait = false, const Table& filter = {}) :
BasicFrame(channel, (queueName.length() + consumerTag.length() + 5 + filter.size())), // size of vars, +1 for each shortstring size, +1 for bools, +2 for deprecated value
BasicFrame(channel, (uint32_t)(queueName.length() + consumerTag.length() + 5 + filter.size())), // size of vars, +1 for each shortstring size, +1 for bools, +2 for deprecated value
_queueName(queueName),
_consumerTag(consumerTag),
_bools(noLocal, noAck, exclusive, noWait),

2
src/basicconsumeokframe.h
View File

@ -43,7 +43,7 @@ public:
* @param consumerTag consumertag specified by client of provided by server
*/
BasicConsumeOKFrame(uint16_t channel, const std::string& consumerTag) :
BasicFrame(channel, consumerTag.length() + 1), // length of string + 1 for encoding of stringsize
BasicFrame(channel, (uint32_t)(consumerTag.length() + 1)), // length of string + 1 for encoding of stringsize
_consumerTag(consumerTag)
{}

8
src/basicdeliverframe.h
View File

@ -13,9 +13,9 @@
* Dependencies
*/
#include "basicframe.h"
#include "../include/stringfield.h"
#include "../include/booleanset.h"
#include "../include/connectionimpl.h"
#include "amqpcpp/stringfield.h"
#include "amqpcpp/booleanset.h"
#include "amqpcpp/connectionimpl.h"
/**
* Set up namespace
@ -87,7 +87,7 @@ public:
* @param routingKey message routing key
*/
BasicDeliverFrame(uint16_t channel, const std::string& consumerTag, uint64_t deliveryTag, bool redelivered = false, const std::string& exchange = "", const std::string& routingKey = "") :
BasicFrame(channel, (consumerTag.length() + exchange.length() + routingKey.length() + 12)),
BasicFrame(channel, (uint32_t)(consumerTag.length() + exchange.length() + routingKey.length() + 12)),
// length of strings + 1 byte per string for stringsize, 8 bytes for uint64_t and 1 for bools
_consumerTag(consumerTag),
_deliveryTag(deliveryTag),

2
src/basicgetframe.h
View File

@ -59,7 +59,7 @@ public:
* @param noAck whether server expects acknowledgements for messages
*/
BasicGetFrame(uint16_t channel, const std::string& queue, bool noAck = false) :
BasicFrame(channel, queue.length() + 4), // 1 for bool, 1 for string size, 2 for deprecated field
BasicFrame(channel, (uint32_t)(queue.length() + 4)), // 1 for bool, 1 for string size, 2 for deprecated field
_queue(queue),
_noAck(noAck)
{}

2
src/basicgetokframe.h
View File

@ -76,7 +76,7 @@ public:
* @param messageCount number of messages in the queue
*/
BasicGetOKFrame(uint16_t channel, uint64_t deliveryTag, bool redelivered, const std::string& exchange, const std::string& routingKey, uint32_t messageCount) :
BasicFrame(channel, (exchange.length() + routingKey.length() + 15)), // string length, +1 for each shortsrting length + 8 (uint64_t) + 4 (uint32_t) + 1 (bool)
BasicFrame(channel, (uint32_t)(exchange.length() + routingKey.length() + 15)), // string length, +1 for each shortsrting length + 8 (uint64_t) + 4 (uint32_t) + 1 (bool)
_deliveryTag(deliveryTag),
_redelivered(redelivered),
_exchange(exchange),

8
src/basicheaderframe.h
View File

@ -13,10 +13,10 @@
* Dependencies
*/
#include "headerframe.h"
#include "../include/metadata.h"
#include "../include/envelope.h"
#include "../include/connectionimpl.h"
#include "../include/deferredconsumerbase.h"
#include "amqpcpp/metadata.h"
#include "amqpcpp/envelope.h"
#include "amqpcpp/connectionimpl.h"
#include "amqpcpp/deferredconsumerbase.h"
/**
* Set up namespace

2
src/basicpublishframe.h
View File

@ -70,7 +70,7 @@ public:
* @param immediate request immediate delivery @default = false
*/
BasicPublishFrame(uint16_t channel, const std::string& exchange = "", const std::string& routingKey = "", bool mandatory = false, bool immediate = false) :
BasicFrame(channel, exchange.length() + routingKey.length() + 5), // 1 extra per string (for the size), 1 for bools, 2 for deprecated field
BasicFrame(channel, (uint32_t)(exchange.length() + routingKey.length() + 5)), // 1 extra per string (for the size), 1 for bools, 2 for deprecated field
_exchange(exchange),
_routingKey(routingKey),
_bools(mandatory, immediate)

2
src/basicreturnframe.h
View File

@ -67,7 +67,7 @@ public:
* @param routingKey message routing key
*/
BasicReturnFrame(uint16_t channel, int16_t replyCode, const std::string& replyText = "", const std::string& exchange = "", const std::string& routingKey = "") :
BasicFrame(channel, replyText.length() + exchange.length() + routingKey.length() + 5), // 3 for each string (extra size byte), 2 for uint16_t
BasicFrame(channel, (uint32_t)(replyText.length() + exchange.length() + routingKey.length() + 5)), // 3 for each string (extra size byte), 2 for uint16_t
_replyCode(replyCode),
_replyText(replyText),
_exchange(exchange),

4
src/bodyframe.h
View File

@ -13,8 +13,8 @@
* Dependencies
*/
#include "extframe.h"
#include "../include/connectionimpl.h"
#include "../include/deferredconsumerbase.h"
#include "amqpcpp/connectionimpl.h"
#include "amqpcpp/deferredconsumerbase.h"
/**
* Set up namespace

2
src/channelcloseframe.h
View File

@ -82,7 +82,7 @@ public:
* @param failingMethod failing method id if applicable
*/
ChannelCloseFrame(uint16_t channel, uint16_t code = 0, const std::string& text = "", uint16_t failingClass = 0, uint16_t failingMethod = 0) :
ChannelFrame(channel, (text.length() + 7)), // sizeof code, failingclass, failingmethod (2byte + 2byte + 2byte) + text length + text length byte
ChannelFrame(channel, (uint32_t)(text.length() + 7)), // sizeof code, failingclass, failingmethod (2byte + 2byte + 2byte) + text length + text length byte
_code(code),
_text(text),
_failingClass(failingClass),

2
src/channelimpl.cpp
View File

@ -481,7 +481,7 @@ bool ChannelImpl::publish(const std::string &exchange, const std::string &routin
uint64_t chunksize = std::min(static_cast<uint64_t>(maxpayload), bytesleft);
// send out a body frame
if (!send(BodyFrame(_id, data + bytessent, chunksize))) return false;
if (!send(BodyFrame(_id, data + bytessent, (uint32_t)chunksize))) return false;
// channel still valid?
if (!monitor.valid()) return false;

2
src/connectioncloseframe.h
View File

@ -82,7 +82,7 @@ public:
* @param failingMethod id of the failing method if applicable
*/
ConnectionCloseFrame(uint16_t code, const std::string &text, uint16_t failingClass = 0, uint16_t failingMethod = 0) :
ConnectionFrame(text.length() + 7), // 1 for extra string byte, 2 for each uint16
ConnectionFrame((uint32_t)(text.length() + 7)), // 1 for extra string byte, 2 for each uint16
_code(code),
_text(text),
_failingClass(failingClass),

4
src/connectionimpl.cpp
View File

@ -125,7 +125,7 @@ uint64_t ConnectionImpl::parse(const Buffer &buffer)
try
{
// try to recognize the frame
ReducedBuffer reduced_buf(buffer, processed);
ReducedBuffer reduced_buf(buffer, (size_t)processed);
ReceivedFrame receivedFrame(reduced_buf, _maxFrame);
// do we have the full frame?
@ -146,7 +146,7 @@ uint64_t ConnectionImpl::parse(const Buffer &buffer)
// have the initial bytes of the header, we already know how much
// data we need for the next frame, otherwise we need at least 7
// bytes for processing the header of the next frame
_expected = receivedFrame.header() ? receivedFrame.totalSize() : 7;
_expected = receivedFrame.header() ? (uint32_t)receivedFrame.totalSize() : 7;
// we're ready for now
return processed;

2
src/connectionopenframe.h
View File

@ -60,7 +60,7 @@ public:
* @param vhost name of virtual host to open
*/
ConnectionOpenFrame(const std::string &vhost) :
ConnectionFrame(vhost.length() + 3), // length of vhost + byte to encode this length + deprecated shortstring size + deprecated bool
ConnectionFrame((uint32_t)(vhost.length() + 3)), // length of vhost + byte to encode this length + deprecated shortstring size + deprecated bool
_vhost(vhost),
_deprecatedCapabilities(""),
_deprecatedInsist()

2
src/connectionsecureframe.h
View File

@ -43,7 +43,7 @@ public:
* @param challenge the challenge
*/
ConnectionSecureFrame(const std::string& challenge) :
ConnectionFrame(challenge.length() + 4), // 4 for the length of the challenge (uint32_t)
ConnectionFrame((uint32_t)(challenge.length() + 4)), // 4 for the length of the challenge (uint32_t)
_challenge(challenge)
{}

2
src/connectionsecureokframe.h
View File

@ -43,7 +43,7 @@ public:
* @param response the challenge response
*/
ConnectionSecureOKFrame(const std::string& response) :
ConnectionFrame(response.length() + 4), //response length + uint32_t for encoding the length
ConnectionFrame((uint32_t)(response.length() + 4)), //response length + uint32_t for encoding the length
_response(response)
{}

2
src/connectionstartframe.h
View File

@ -81,7 +81,7 @@ public:
* @param locales available locales
*/
ConnectionStartFrame(uint8_t major, uint8_t minor, const Table& properties, const std::string& mechanisms, const std::string& locales) :
ConnectionFrame((properties.size() + mechanisms.length() + locales.length() + 10)), // 4 for each longstring (size-uint32), 2 major/minor
ConnectionFrame((uint32_t)(properties.size() + mechanisms.length() + locales.length() + 10)), // 4 for each longstring (size-uint32), 2 major/minor
_major(major),
_minor(minor),
_properties(properties),

2
src/connectionstartokframe.h
View File

@ -82,7 +82,7 @@ public:
* @param locale selected locale.
*/
ConnectionStartOKFrame(const Table& properties, const std::string& mechanism, const std::string& response, const std::string& locale) :
ConnectionFrame((properties.size() + mechanism.length() + response.length() + locale.length() + 6)), // 1 byte extra per shortstring, 4 per longstring
ConnectionFrame((uint32_t)(properties.size() + mechanism.length() + response.length() + locale.length() + 6)), // 1 byte extra per shortstring, 4 per longstring
_properties(properties),
_mechanism(mechanism),
_response(response),

2
src/deferredconsumerbase.cpp
View File

@ -10,7 +10,7 @@
/**
* Dependencies
*/
#include "../include/deferredconsumerbase.h"
#include "amqpcpp/deferredconsumerbase.h"
#include "basicdeliverframe.h"
#include "basicgetokframe.h"
#include "basicheaderframe.h"

2
src/exchangebindframe.h
View File

@ -95,7 +95,7 @@ public:
* @param arguments
*/
ExchangeBindFrame(uint16_t channel, const std::string &destination, const std::string &source, const std::string &routingKey, bool noWait, const Table &arguments) :
ExchangeFrame(channel, (destination.length() + source.length() + routingKey.length() + arguments.size() + 6)), // 1 for each string, 1 for booleanset, 2 for deprecated field
ExchangeFrame(channel, (uint32_t)(destination.length() + source.length() + routingKey.length() + arguments.size() + 6)), // 1 for each string, 1 for booleanset, 2 for deprecated field
_destination(destination),
_source(source),
_routingKey(routingKey),

2
src/exchangedeclareframe.h
View File

@ -81,7 +81,7 @@ public:
* @param arguments additional arguments
*/
ExchangeDeclareFrame(uint16_t channel, const std::string& name, const std::string& type, bool passive, bool durable, bool noWait, const Table& arguments) :
ExchangeFrame(channel, (name.length() + type.length() + arguments.size() + 5)), // size of name, type and arguments + 1 (all booleans are stored in 1 byte) + 2 (deprecated short) + 2 (string sizes)
ExchangeFrame(channel, (uint32_t)(name.length() + type.length() + arguments.size() + 5)), // size of name, type and arguments + 1 (all booleans are stored in 1 byte) + 2 (deprecated short) + 2 (string sizes)
_name(name),
_type(type),
_bools(passive, durable, false, false, noWait),

2
src/exchangedeleteframe.h
View File

@ -73,7 +73,7 @@ public:
* @param bool noWait Do not wait for a response
*/
ExchangeDeleteFrame(uint16_t channel, const std::string& name, bool ifUnused = false, bool noWait = false) :
ExchangeFrame(channel, name.length() + 4), // length of the name, 1 byte for encoding this length, 1 for bools, 2 for deprecated short
ExchangeFrame(channel, (uint32_t)(name.length() + 4)), // length of the name, 1 byte for encoding this length, 1 for bools, 2 for deprecated short
_name(name),
_bools(ifUnused, noWait)
{}

2
src/exchangeunbindframe.h
View File

@ -95,7 +95,7 @@ public:
* @param arguments
*/
ExchangeUnbindFrame(uint16_t channel, const std::string &destination, const std::string &source, const std::string &routingKey, bool noWait, const Table &arguments) :
ExchangeFrame(channel, (destination.length() + source.length() + routingKey.length() + arguments.size() + 6)), // 1 for each string, 1 for booleanset, 2 for deprecated field
ExchangeFrame(channel, (uint32_t)(destination.length() + source.length() + routingKey.length() + arguments.size() + 6)), // 1 for each string, 1 for booleanset, 2 for deprecated field
_destination(destination),
_source(source),
_routingKey(routingKey),

4
src/extframe.h
View File

@ -19,8 +19,8 @@
/**
* Dependencies
*/
#include "../include/frame.h"
#include "../include/receivedframe.h"
#include "amqpcpp/frame.h"
#include "amqpcpp/receivedframe.h"
/**
* Set up namespace

2
src/framecheck.h
View File

@ -50,7 +50,7 @@ public:
virtual ~FrameCheck()
{
// update the number of bytes to skip
_frame->_skip += _size;
_frame->_skip += (uint32_t)_size;
}
};

99
src/includes.h
View File

@ -9,7 +9,7 @@
// c and c++ dependencies
#include <stdlib.h>
#include <string.h>
#include <string.h> // TODO cstring
#include <stdint.h>
#include <string>
#include <memory>
@ -21,67 +21,69 @@
#include <unordered_map>
#include <vector>
#include <queue>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <unistd.h>
#include <netinet/tcp.h>
#include <sys/types.h> // TODO is this needed
#include <functional>
#include <stdexcept>
// TODO make this nice
#ifdef _MSC_VER
//not #if defined(_WIN32) || defined(_WIN64) because we have strncasecmp in mingw
#define strncasecmp _strnicmp
#define strcasecmp _stricmp
#endif
// forward declarations
#include "../include/classes.h"
#include "amqpcpp/classes.h"
// utility classes
#include "../include/endian.h"
#include "../include/buffer.h"
#include "../include/bytebuffer.h"
#include "../include/receivedframe.h"
#include "../include/outbuffer.h"
#include "../include/copiedbuffer.h"
#include "../include/watchable.h"
#include "../include/monitor.h"
#include "../include/tcpdefines.h"
#include "amqpcpp/endian.h"
#include "amqpcpp/buffer.h"
#include "amqpcpp/bytebuffer.h"
#include "amqpcpp/receivedframe.h"
#include "amqpcpp/outbuffer.h"
#include "amqpcpp/copiedbuffer.h"
#include "amqpcpp/watchable.h"
#include "amqpcpp/monitor.h"
// amqp types
#include "../include/field.h"
#include "../include/numericfield.h"
#include "../include/decimalfield.h"
#include "../include/stringfield.h"
#include "../include/booleanset.h"
#include "../include/fieldproxy.h"
#include "../include/table.h"
#include "../include/array.h"
#include "amqpcpp/field.h"
#include "amqpcpp/numericfield.h"
#include "amqpcpp/decimalfield.h"
#include "amqpcpp/stringfield.h"
#include "amqpcpp/booleanset.h"
#include "amqpcpp/fieldproxy.h"
#include "amqpcpp/table.h"
#include "amqpcpp/array.h"
// envelope for publishing and consuming
#include "../include/metadata.h"
#include "../include/envelope.h"
#include "../include/message.h"
#include "amqpcpp/metadata.h"
#include "amqpcpp/envelope.h"
#include "amqpcpp/message.h"
// mid level includes
#include "../include/exchangetype.h"
#include "../include/flags.h"
#include "../include/callbacks.h"
#include "../include/deferred.h"
#include "../include/deferredconsumer.h"
#include "../include/deferredqueue.h"
#include "../include/deferreddelete.h"
#include "../include/deferredcancel.h"
#include "../include/deferredget.h"
#include "../include/channelimpl.h"
#include "../include/channel.h"
#include "../include/login.h"
#include "../include/address.h"
#include "../include/connectionhandler.h"
#include "../include/connectionimpl.h"
#include "../include/connection.h"
#include "../include/tcphandler.h"
#include "../include/tcpconnection.h"
#include "amqpcpp/exchangetype.h"
#include "amqpcpp/flags.h"
#include "amqpcpp/callbacks.h"
#include "amqpcpp/deferred.h"
#include "amqpcpp/deferredconsumer.h"
#include "amqpcpp/deferredqueue.h"
#include "amqpcpp/deferreddelete.h"
#include "amqpcpp/deferredcancel.h"
#include "amqpcpp/deferredget.h"
#include "amqpcpp/channelimpl.h"
#include "amqpcpp/channel.h"
#include "amqpcpp/login.h"
#include "amqpcpp/address.h"
#include "amqpcpp/connectionhandler.h"
#include "amqpcpp/connectionimpl.h"
#include "amqpcpp/connection.h"
// classes that are very commonly used
#include "../include/exception.h"
#include "../include/protocolexception.h"
#include "../include/frame.h"
#include "amqpcpp/exception.h"
#include "amqpcpp/protocolexception.h"
#include "amqpcpp/frame.h"
#include "extframe.h"
#include "methodframe.h"
#include "headerframe.h"
@ -91,6 +93,5 @@
#include "queueframe.h"
#include "basicframe.h"
#include "transactionframe.h"
#include "addressinfo.h"

12
src/linux_tcp/CMakeLists.txt Normal file
View File

@ -0,0 +1,12 @@
add_sources(
addressinfo.h
includes.h
pipe.h
tcpclosed.h
tcpconnected.h
tcpconnection.cpp
tcpinbuffer.h
tcpoutbuffer.h
tcpresolver.h
tcpstate.h
)

0
src/addressinfo.h → src/linux_tcp/addressinfo.h
View File

Some files were not shown because too many files have changed in this diff Show More