#include <Socket_ZMQ4.hpp>

Inheritance diagram for opentxs::OTSocket_ZMQ_4:

Collaboration diagram for opentxs::OTSocket_ZMQ_4:

Classes
class	ZMQ4

Public Member Functions
EXPORT	OTSocket_ZMQ_4 ()

EXPORT	~OTSocket_ZMQ_4 ()

EXPORT bool	NewContext ()

EXPORT bool	RemakeSocket (const bool bNewContext=false)

EXPORT bool	Connect ()

EXPORT bool	Listen ()

EXPORT bool	Connect (const OTString &strConnectPath)

EXPORT bool	Listen (const OTString &strBindingPath)

EXPORT bool	Send (const OTASCIIArmor &ascEnvelope)

EXPORT bool	Send (const OTASCIIArmor &ascEnvelope, const OTString &strConnectPath)

EXPORT bool	Receive (OTString &strServerReply)

Public Member Functions inherited from opentxs::OTSocket
virtual	~OTSocket ()

EXPORT std::mutex *	GetMutex ()

EXPORT bool	Init (const Defaults &defaults)

EXPORT bool	Init (const Defaults &defaults, OTSettings *pSettings)

EXPORT bool	IsInitialized () const

EXPORT bool	HasContext () const

EXPORT bool	IsConnected () const

EXPORT bool	IsListening () const

EXPORT const OTString &	GetConnectPath () const

EXPORT const OTString &	GetBindingPath () const

Additional Inherited Members
Protected Member Functions inherited from opentxs::OTSocket
	OTSocket ()

Protected Attributes inherited from opentxs::OTSocket
int64_t	m_lLatencySendMs

int32_t	m_nLatencySendNoTries

int64_t	m_lLatencyReceiveMs

int32_t	m_nLatencyReceiveNoTries

int64_t	m_lLatencyDelayAfter

bool	m_bIsBlocking

bool	m_bInitialized

bool	m_HasContext

bool	m_bConnected

bool	m_bListening

OTString	m_strConnectPath

OTString	m_strBindingPath

OTASCIIArmor	m_ascLastMsgSent

Detailed Description

Definition at line 141 of file Socket_ZMQ4.hpp.

Constructor & Destructor Documentation

opentxs::OTSocket_ZMQ_4::OTSocket_ZMQ_4 ( )

Definition at line 164 of file Socket_ZMQ4.cpp.

     : m_pzmq(new ZMQ4())
 {
 }

opentxs::OTSocket_ZMQ_4::~OTSocket_ZMQ_4 ( )

Definition at line 169 of file Socket_ZMQ4.cpp.

 {
     CloseSocket();
     delete (m_pzmq);
 }

Member Function Documentation

bool opentxs::OTSocket_ZMQ_4::Connect ( )

virtual

Implements opentxs::OTSocket.

Definition at line 274 of file Socket_ZMQ4.cpp.

 {
     if (!m_bInitialized) {
         OT_FAIL;
     }
     if (!m_HasContext) {
         OT_FAIL;
     }
 
     if (nullptr == m_pzmq->context_zmq) {
         OTLog::vError("%s: Error: %s must exist to Listen!\n", __FUNCTION__,
                       "m_pzmq->context_zmq");
         OT_FAIL;
     }
     if (true == m_bListening) {
         OTLog::vError("%s: Error: Must not be Listening, to Connect!\n",
                       __FUNCTION__);
         OT_FAIL;
     }
 
     if (!m_strConnectPath.Exists()) {
         OT_FAIL;
     }
 
     if (!NewSocket(true)) return false; // NewSocket(true), Request Socket.
 
     try {
         m_pzmq->socket_zmq->connect(m_strConnectPath.Get());
     }
     catch (std::exception& e) {
         OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__, e.what());
         OT_FAIL;
     }
 
     m_bConnected = true;
     return true;
 }

bool opentxs::OTSocket_ZMQ_4::Connect ( const OTString & strConnectPath )

virtual

Implements opentxs::OTSocket.

Definition at line 353 of file Socket_ZMQ4.cpp.

 {
     if (!strConnectPath.Exists()) {
         OTLog::vError("%s: Error: %s dosn't exist!\n", __FUNCTION__,
                       "strConnectPath");
         OT_FAIL;
     }
     if (5 > strConnectPath.GetLength()) {
         OTLog::vError("%s: Error: %s is too short!\n", __FUNCTION__,
                       "strConnectPath");
         OT_FAIL;
     }
 
     m_strConnectPath = strConnectPath; // set the connection path.
 
     return (Connect());
 }

bool opentxs::OTSocket_ZMQ_4::Listen ( )

virtual

Implements opentxs::OTSocket.

Definition at line 312 of file Socket_ZMQ4.cpp.

 {
     if (!m_bInitialized) {
         OT_FAIL;
     }
     if (!m_HasContext) {
         OT_FAIL;
     }
 
     if (nullptr == m_pzmq->context_zmq) {
         OTLog::vError("%s: Error: %s must exist to Listen!\n", __FUNCTION__,
                       "m_pzmq->context_zmq");
         OT_FAIL;
     }
     if (true == m_bConnected) {
         OTLog::vError("%s: Error: Must not be Connected, to Listen!\n",
                       __FUNCTION__);
         OT_FAIL;
     }
 
     if (!m_strBindingPath.Exists()) {
         OT_FAIL;
     }
 
     if (!NewSocket(false)) return false; // NewSocket(false), Responce Socket.
 
     try {
         m_pzmq->socket_zmq->bind(m_strBindingPath.Get()); // since
                                                           // m_strBindingPath
                                                           // was checked and set
                                                           // above.
     }
     catch (std::exception& e) {
         OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__, e.what());
         OT_FAIL;
     }
 
     m_bListening = true;
     return true;
 }

bool opentxs::OTSocket_ZMQ_4::Listen ( const OTString & strBindingPath )

virtual

Implements opentxs::OTSocket.

Definition at line 371 of file Socket_ZMQ4.cpp.

 {
     if (!strBindingPath.Exists()) {
         OTLog::vError("%s: Error: %s dosn't exist!\n", __FUNCTION__,
                       "strBindingPath");
         OT_FAIL;
     }
     if (5 > strBindingPath.GetLength()) {
         OTLog::vError("%s: Error: %s is too short!\n", __FUNCTION__,
                       "strBindingPath");
         OT_FAIL;
     }
 
     m_strBindingPath = strBindingPath;
 
     return (Listen());
 }

bool opentxs::OTSocket_ZMQ_4::NewContext ( )

virtual

Implements opentxs::OTSocket.

Definition at line 229 of file Socket_ZMQ4.cpp.

 {
     if (!m_bInitialized) return false;
 
     m_HasContext = false;
 
     if (!CloseSocket(true)) return false;
 
     if (nullptr != m_pzmq->context_zmq) zmq_term(m_pzmq->context_zmq);
     if (nullptr != m_pzmq->context_zmq) delete m_pzmq->context_zmq;
     m_pzmq->context_zmq = nullptr;
 
     try {
         m_pzmq->context_zmq = new zmq::context_t(
             1, 31); // Threads, Max Sockets. (31 is a sane default).
     }
     catch (std::exception& e) {
         OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__, e.what());
         OT_FAIL;
     }
 
     m_HasContext = true;
     return true;
 }

bool opentxs::OTSocket_ZMQ_4::Receive ( OTString & strServerReply )

virtual

Implements opentxs::OTSocket.

Definition at line 521 of file Socket_ZMQ4.cpp.

 {
     if (!m_bInitialized) {
         OT_FAIL;
     }
     if (!m_HasContext) {
         OT_FAIL;
     }
     if (nullptr == m_pzmq->context_zmq) {
         OTLog::vError("%s: Error: %s must exist to Receive!\n", __FUNCTION__,
                       "m_pzmq->context_zmq");
         OT_FAIL;
     }
 
     if (!m_bConnected && !m_bListening) return false;
     if (m_bConnected && m_bListening) return false;
     if (nullptr == m_pzmq->socket_zmq) {
         OTLog::vError("%s: Error: %s must exist to Receive!\n", __FUNCTION__,
                       "m_pzmq->socket_zmq");
         OT_FAIL;
     }
 
     // -----------------------------------
     const int64_t lLatencyRecvMilliSec = m_lLatencyReceiveMs;
 
     //  Get the reply.
     zmq::message_t zmq_message;
 
     bool bSuccessReceiving = false;
 
     // If failure receiving, re-tries 2 times, with 4000 ms max delay between
     // each (Doubling every time.)
     //
     if (m_bIsBlocking) {
         try {
             bSuccessReceiving =
                 m_pzmq->socket_zmq->recv(&zmq_message); // Blocking.
         }
         catch (std::exception& e) {
             OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__,
                           e.what());
             OT_FAIL;
         }
     }
     else // not blocking
     {
         int64_t lDoubling = lLatencyRecvMilliSec;
         int32_t nReceiveTries = m_nLatencyReceiveNoTries;
         bool expect_reply = true;
         while (expect_reply) {
             //  Poll socket for a reply, with timeout
             zmq::pollitem_t items[] = {{*m_pzmq->socket_zmq, 0, ZMQ_POLLIN, 0}};
 
             int nPoll = 0;
             try {
                 nPoll = zmq::poll(&items[0], 1, static_cast<long>(lDoubling));
             }
             catch (std::exception& e) {
                 OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__,
                               e.what());
                 OT_FAIL;
             }
 
             lDoubling *= 2;
 
             //  If we got a reply, process it
             if (items[0].revents & ZMQ_POLLIN) {
                 try {
                     bSuccessReceiving = m_pzmq->socket_zmq->recv(
                         &zmq_message,
                         ZMQ_NOBLOCK); // <=========== RECEIVE ===============
                 }
                 catch (std::exception& e) {
                     OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__,
                                   e.what());
                     OT_FAIL;
                 }
                 OTLog::SleepMilliseconds(1);
 
                 if (!bSuccessReceiving) {
                     if (!HandleReceivingError()) expect_reply = false;
                 }
                 else
                     break; // (Success -- we're done in this loop.)
             }
             else if (nReceiveTries == 0) {
                 // OTLog::Error("OTSocket::Receive: no message.\n");
                 expect_reply = false;
                 break;
             }
             else if ((-1) == nPoll) // error.
             {
                 if (!HandlePollingError()) expect_reply = false;
             }
 
             --nReceiveTries;
         }
     }
 
     if (bSuccessReceiving && (zmq_message.size() > 0))
         strServerReply.MemSet(static_cast<const char*>(zmq_message.data()),
                               static_cast<uint32_t>(zmq_message.size()));
 
     return (bSuccessReceiving && (zmq_message.size() > 0));
 }

bool opentxs::OTSocket_ZMQ_4::RemakeSocket ( const bool bNewContext = false )

virtual

Implements opentxs::OTSocket.

Definition at line 254 of file Socket_ZMQ4.cpp.

 {
 
     if (!m_bInitialized) return false;
     if (!m_HasContext) return false;
 
     if (!m_bConnected || !m_bListening) return false;
     if (m_bConnected && m_bListening) return false;
 
     bool bConnected = m_bConnected;
     bool bListening = m_bListening;
 
     if (bNewContext) NewContext();
 
     if (bConnected) return Connect();
     if (bListening) return Listen();
 
     return false;
 }

bool opentxs::OTSocket_ZMQ_4::Send ( const OTASCIIArmor & ascEnvelope )

virtual

Implements opentxs::OTSocket.

Definition at line 389 of file Socket_ZMQ4.cpp.

 {
     if (!m_bInitialized) {
         OT_FAIL;
     }
 
     if (0 >= ascEnvelope.GetLength()) {
         OTLog::vError("%s: Error: %s is zero length!\n", __FUNCTION__,
                       "ascEnvelope");
         OT_FAIL;
     }
     m_ascLastMsgSent.Set(ascEnvelope); // In case we need to re-send.
 
     if (!m_HasContext) {
         OT_FAIL;
     }
     if (nullptr == m_pzmq->context_zmq) {
         OTLog::vError("%s: Error: %s must exist to Send!\n", __FUNCTION__,
                       "m_pzmq->context_zmq");
         OT_FAIL;
     }
 
     if (!m_bConnected && !m_bListening) return false;
     if (m_bConnected && m_bListening) return false;
     if (nullptr == m_pzmq->socket_zmq) {
         OTLog::vError("%s: Error: %s must exist to Send!\n", __FUNCTION__,
                       "m_pzmq->socket_zmq");
         OT_FAIL;
     }
 
     // -----------------------------------
     const int64_t lLatencySendMilliSec = m_lLatencySendMs;
 
     zmq::message_t zmq_message(ascEnvelope.GetLength());
     memcpy((void*)zmq_message.data(), ascEnvelope.Get(),
            ascEnvelope.GetLength());
 
     bool bSuccessSending = false;
 
     if (m_bIsBlocking) {
         try {
             bSuccessSending =
                 m_pzmq->socket_zmq->send(zmq_message); // Blocking.
         }
         catch (std::exception& e) {
             OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__,
                           e.what());
             OT_FAIL;
         }
     }
     else // not blocking
     {
         int32_t nSendTries = m_nLatencySendNoTries;
         int64_t lDoubling = lLatencySendMilliSec;
         bool bKeepTrying = true;
 
         while (bKeepTrying && (nSendTries > 0)) {
             zmq::pollitem_t items[] = {
                 {(*m_pzmq->socket_zmq), 0, ZMQ_POLLOUT, 0}};
 
             int nPoll = 0;
             try {
                 nPoll =
                     zmq::poll(&items[0], 1,
                               static_cast<long>(lDoubling)); // ZMQ_POLLOUT, 1
                                                              // item, timeout
                                                              // (milliseconds)
             }
             catch (std::exception& e) {
                 OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__,
                               e.what());
                 OT_FAIL;
             }
 
             lDoubling *= 2;
 
             if (items[0].revents & ZMQ_POLLOUT) {
                 try {
                     bSuccessSending = m_pzmq->socket_zmq->send(
                         zmq_message,
                         ZMQ_NOBLOCK); // <=========== SEND ===============
                 }
                 catch (std::exception& e) {
                     OTLog::vError("%s: Exception Caught: %s \n", __FUNCTION__,
                                   e.what());
                     OT_FAIL;
                 }
 
                 OTLog::SleepMilliseconds(1);
 
                 if (!bSuccessSending) {
                     if (!HandleSendingError()) bKeepTrying = false;
                 }
                 else
                     break;            // (Success -- we're done in this loop.)
             }
             else if ((-1) == nPoll) // error.
             {
                 if (!HandlePollingError()) bKeepTrying = false;
             }
 
             --nSendTries;
         }
     }
     /*
     Normally, we try to send...
     If the send fails, we wait X ms and then try again (Y times).
 
     BUT -- what if the failure was an errno==EAGAIN ?
     In that case, it's not a REAL failure, but rather, a "failure right now, try
     again in a sec."
     */
 
     if (bSuccessSending)
         OTLog::SleepMilliseconds(m_lLatencyDelayAfter > 0 ? m_lLatencyDelayAfter
                                                           : 1);
 
     return bSuccessSending;
 }

bool opentxs::OTSocket_ZMQ_4::Send	(	const OTASCIIArmor &	ascEnvelope,
		const OTString &	strConnectPath
	)

virtual

Implements opentxs::OTSocket.

Definition at line 509 of file Socket_ZMQ4.cpp.

 {
     const bool bNewPath = m_strConnectPath.Compare(strConnectPath);
 
     if (!bNewPath) Connect(strConnectPath);
 
     if (!m_bConnected) OT_FAIL;
 
     return Send(ascEnvelope);
 }

The documentation for this class was generated from the following files:

/usr/local/src/opentxs/include/opentxs/ext/Socket_ZMQ4.hpp
/usr/local/src/opentxs/src/ext/Socket_ZMQ4.cpp

Classes

Public Member Functions

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation