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

Use address to pass the option and add some extra options for sorting ips.

This commit is contained in:
aljar 2020-11-16 16:30:29 +01:00
parent fe1538e8ae
commit f9d85f5d01
5 changed files with 283 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
src/linux_tcp/address2str.h Normal file
View File

@ -0,0 +1,108 @@
/**
* Address2Str.h
*
* Helper class to get a stringed version of an address obtained from get_addrinfo
*
* @author Aljar Meesters <aljar.meesters@copernica.com
* @copyright 2020 Copernica BV
*/
/**
* Include guard
*/
#pragma once
/**
* Dependencies
*/
#include <sys/socket.h>
#include <arpa/inet.h>
/**
* Opening namespace
*/
namespace AMQP {
/**
* class implemenation
*/
class Address2str
{
private:
/**
* The address as string
*/
char *_buffer = nullptr;
public:
/**
* Constructor
* @param struct addrinfo *
*/
Address2str(struct addrinfo *info)
{
// Switch on family
switch(info->ai_family)
{
// If we are ip4
case AF_INET:
{
// ugly cast
struct sockaddr_in *addr_in = (struct sockaddr_in *)(info->ai_addr);
// create the buffer
_buffer = new char[INET_ADDRSTRLEN];
// get the info in our buffer
inet_ntop(AF_INET, &(addr_in->sin_addr), _buffer, INET_ADDRSTRLEN);
// done
break;
}
// if we are ipv6
case AF_INET6:
{
// ugly cast
struct sockaddr_in6 *addr_in6 = (struct sockaddr_in6 *)(info->ai_addr);
// crate buffer
_buffer = new char[INET6_ADDRSTRLEN];
// get the info in our buffer
inet_ntop(AF_INET6, &(addr_in6->sin6_addr), _buffer, INET6_ADDRSTRLEN);
// done
break;
}
default:
// If it is not ipv4 or ipv6 we don't set it.
break;
}
}
/**
* Destruction
*/
virtual ~Address2str()
{
// if we have a buffer, we should free it.
if (_buffer) delete[](_buffer);
}
/**
* get the char
* @return char*
*/
const char *toChar() const
{
// expose the buffer.
return _buffer;
}
};
/**
* ending of namespace
*/
}

112
src/linux_tcp/addressinfo.h
View File

@ -11,6 +11,8 @@
* Dependencies
*/
#include <random>
#include "connectionorder.h"
#include "address2str.h"
/**
* Include guard
@ -35,14 +37,103 @@ private:
*/
std::vector<struct addrinfo *> _v;
/**
* Helper function to order the vector of addrinfo based on the ordering received
* @param order
*/
void reorder(ConnectionOrder::Order order)
{
// witch on order
switch (order)
{
// Do we want to have a random order of the addresses?
// This may be useful since getaddrinfo is sorting the addresses on proximity
// (e.g. https://lists.debian.org/debian-glibc/2007/09/msg00347.html),
// which may break loadbalancing..
case ConnectionOrder::Order::random:
{
// create a random device for the seed of the random number generator
std::random_device rd;
// Create the generator
std::mt19937 gen(rd());
// shuffle the vector.
std::shuffle(_v.begin(), _v.end(), gen);
// done
break;
}
// do we want to sort in ascending order
case ConnectionOrder::Order::ascending:
{
std::sort(_v.begin(), _v.end(), []
(struct addrinfo * v1, struct addrinfo * v2) -> bool
{
// get the addresses
Address2str addr1(v1);
Address2str addr2(v2);
// if addr1 doesn't have a proper address it should go to the
// back. Same holds for addr2
if (addr1.toChar() == nullptr) return false;
if (addr2.toChar() == nullptr) return true;
// make the comparison based on string comparison
return strcmp(addr1.toChar(), addr2.toChar()) < 0;
});
// done
break;
}
// do we want to sort in descending order
case ConnectionOrder::Order::descending:
{
std::sort(_v.begin(), _v.end(), []
(struct addrinfo * v1, struct addrinfo * v2) -> bool
{
// get the addresses
Address2str addr1(v1);
Address2str addr2(v2);
// if addr1 doesn't have a proper address it should go to the
// back. Same holds for addr2
if (addr1.toChar() == nullptr) return false;
if (addr2.toChar() == nullptr) return true;
// make the comparison based on string comparison
return strcmp(addr1.toChar(), addr2.toChar()) > 0;
});
// done
break;
}
// de we want to have reverse ordering of proximity
case ConnectionOrder::Order::reverse:
{
std::reverse(_v.begin(), _v.end());
// done
break;
}
default:
// nothing to do, just default behaviour
break;
}
}
public:
/**
* Constructor
* @param hostname
* @param port
* @param randomOrder
* @param order
*/
AddressInfo(const char *hostname, uint16_t port = 5672, bool randomOrder = false)
AddressInfo(const char *hostname, uint16_t port = 5672, ConnectionOrder::Order order = ConnectionOrder::Order::standard)
{
// store portnumber in buffer
auto portnumber = std::to_string(port);
@ -70,21 +161,8 @@ public:
_v.push_back(current);
}
// Do we want to have a random order of the addresses?
// This may be useful since getaddrinfo is sorting the addresses on proximity
// (e.g. https://lists.debian.org/debian-glibc/2007/09/msg00347.html),
// which may break loadbalancing..
if (randomOrder)
{
// create a random device for the seed of the random number generator
std::random_device rd;
// Create the generator
std::mt19937 gen(rd());
// shuffle the vector.
std::shuffle(_v.begin(), _v.end(), gen);
}
// Order the vector based on the provided ordering
reorder(order);
}
/**

73
src/linux_tcp/connectionorder.h Normal file
View File

@ -0,0 +1,73 @@
/**
* ConnectionOrder.h
*
* Class that give info on how we want to sellect the connection from a list of IPs
*
* @author Aljar Meesters <aljar.meesters@copernica.com>
* @copyright 2020 Copernica BV
*/
/**
* Include guard
*/
#pragma once
/**
* Setup namespace
*/
namespace AMQP
{
/**
* Class implementation
*/
class ConnectionOrder
{
public:
/**
* Enum class holding the orders we support
* - standard: what is used by getaddrinfo, which is proximity based
* - reverse: reverse the standard order
* - random: random order
* - ascending: try the smallest IP address first
* - descending: try the largest IP address first
* @var enum Order
*/
enum class Order { standard, reverse, random, ascending, descending};
private:
/**
* The order for the connection
* @var Order
*/
Order _order;
public:
/**
* Constructor
* @var order
*/
ConnectionOrder (const char *order) : _order(Order::standard)
{
// Set the orders based on the string
if (strcmp(order, "random") == 0) _order = Order::random;
else if (strcmp(order, "ascending") == 0 || strcmp(order, "asc") == 0) _order = Order::ascending;
else if (strcmp(order, "descending") == 0 || strcmp(order, "desc") == 0 ) _order = Order::descending;
// If we don't ave a match we fall back to standard, which was set as default
}
/**
* Get the order
* @return Order
*/
Order order()
{
return _order;
}
};
/**
* End of namespace
*/
}

2
src/linux_tcp/tcpconnection.cpp
View File

@ -26,7 +26,7 @@ namespace AMQP {
*/
TcpConnection::TcpConnection(TcpHandler *handler, const Address &address) :
_handler(handler),
_state(new TcpResolver(this, address.hostname(), address.port(), address.secure(), address.option("connectTimeout", 5), address.option("randomConnect", false))),
_state(new TcpResolver(this, address.hostname(), address.port(), address.secure(), address.option("connectTimeout", 5), ConnectionOrder(address.option("connectionOrder")).order())),
_connection(this, address.login(), address.vhost())
{
// tell the handler

12
src/linux_tcp/tcpresolver.h
View File

@ -86,10 +86,10 @@ private:
std::thread _thread;
/**
* Do we want to have random ordering
* How should the addresses be ordered when we want to connect
* @var bool
*/
bool _random;
ConnectionOrder::Order _order;
/**
@ -104,7 +104,7 @@ private:
if (_secure && !OpenSSL::valid()) throw std::runtime_error("Secure connection cannot be established: libssl.so cannot be loaded");
// get address info
AddressInfo addresses(_hostname.data(), _port, _random);
AddressInfo addresses(_hostname.data(), _port, _order);
// the pollfd structure, needed for poll()
pollfd fd;
@ -194,15 +194,15 @@ public:
* @param portnumber The portnumber for the lookup
* @param secure Do we need a secure tls connection when ready?
* @param timeout timeout per connection attempt
* @param random Do we want to have random oredering of the address of the host to connect to
* @param order How should we oreder the addresses of the host to connect to
*/
TcpResolver(TcpParent *parent, std::string hostname, uint16_t port, bool secure, int timeout, bool random) :
TcpResolver(TcpParent *parent, std::string hostname, uint16_t port, bool secure, int timeout, ConnectionOrder::Order order) :
TcpExtState(parent),
_hostname(std::move(hostname)),
_secure(secure),
_port(port),
_timeout(timeout),
_random(random)
_order(order)
{
// tell the event loop to monitor the filedescriptor of the pipe
parent->onIdle(this, _pipe.in(), readable);