kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

+add utf-8 validation code, not hooked up properly yet

Browse Source 
+ws autobahn / Add code to test websocket client compliance with the autobahn test-suite
+Ping received with a payload too large (> 125 bytes) trigger a connection closure
+cobra / add tracking about published messages
+cobra / publish returns a message id, that can be used when
+cobra / new message type in the message received handler when publish/ok is received (can be used to implement an ack system).
This commit is contained in:
Benjamin Sergeant
2019-08-31 16:46:44 -07:00
parent a95fcbbdbf
commit b5b0de2083
16 changed files with 375 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
ws/ixcobra/IXCobraConnection.cpp
View File

@ -18,6 +18,7 @@
namespace ix
{
TrafficTrackerCallback CobraConnection::_trafficTrackerCallback = nullptr;
PublishTrackerCallback CobraConnection::_publishTrackerCallback = nullptr;
constexpr size_t CobraConnection::kQueueMaxSize;
CobraConnection::CobraConnection() :
@ -56,6 +57,24 @@ namespace ix
}
}
void CobraConnection::setPublishTrackerCallback(const PublishTrackerCallback& callback)
{
_publishTrackerCallback = callback;
}
void CobraConnection::resetPublishTrackerCallback()
{
setPublishTrackerCallback(nullptr);
}
void CobraConnection::invokePublishTrackerCallback(bool sent, bool acked)
{
if (_publishTrackerCallback)
{
_publishTrackerCallback(sent, acked);
}
}
void CobraConnection::setEventCallback(const EventCallback& eventCallback)
{
std::lock_guard<std::mutex> lock(_eventCallbackMutex);
@ -63,19 +82,20 @@ namespace ix
}
void CobraConnection::invokeEventCallback(ix::CobraConnectionEventType eventType,
const std::string& errorMsg,
const WebSocketHttpHeaders& headers,
const std::string& subscriptionId)
const std::string& errorMsg,
const WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId)
{
std::lock_guard<std::mutex> lock(_eventCallbackMutex);
if (_eventCallback)
{
_eventCallback(eventType, errorMsg, headers, subscriptionId);
_eventCallback(eventType, errorMsg, headers, subscriptionId, msgId);
}
}
void CobraConnection::invokeErrorCallback(const std::string& errorMsg,
const std::string& serializedPdu)
const std::string& serializedPdu)
{
std::stringstream ss;
ss << errorMsg << " : received pdu => " << serializedPdu;
@ -178,6 +198,17 @@ namespace ix
{
invokeErrorCallback("Unsubscription error", msg->str);
}
else if (action == "rtm/publish/ok")
{
if (!handlePublishResponse(data))
{
invokeErrorCallback("Error processing publish response", msg->str);
}
}
else if (action == "rtm/publish/error")
{
invokeErrorCallback("Publish error", msg->str);
}
else
{
invokeErrorCallback("Un-handled message type", msg->str);
@ -374,6 +405,24 @@ namespace ix
return true;
}
bool CobraConnection::handlePublishResponse(const Json::Value& pdu)
{
if (!pdu.isMember("id")) return false;
Json::Value id = pdu["id"];
if (!id.isUInt64()) return false;
uint64_t msgId = id.asUInt64();
invokeEventCallback(ix::CobraConnection_EventType_Published,
std::string(), WebSocketHttpHeaders(),
std::string(), msgId);
invokePublishTrackerCallback(false, true);
return true;
}
bool CobraConnection::connect()
{
_webSocket->start();
@ -399,9 +448,11 @@ namespace ix
//
// publish is not thread safe as we are trying to reuse some Json objects.
//
bool CobraConnection::publish(const Json::Value& channels,
const Json::Value& msg)
CobraConnection::MsgId CobraConnection::publish(const Json::Value& channels,
const Json::Value& msg)
{
invokePublishTrackerCallback(true, false);
_body["channels"] = channels;
_body["message"] = msg;
_pdu["body"] = _body;
@ -412,7 +463,7 @@ namespace ix
if (_publishMode == CobraConnection_PublishMode_Batch)
{
enqueue(serializedJson);
return true;
return _id - 1;
}
//
@ -421,14 +472,14 @@ namespace ix
//
if (_authenticated && publishMessage(serializedJson))
{
return true;
return _id - 1;
}
else // Or else we enqueue
// Slow code path is when we haven't connected yet (startup),
// or when the connection drops for some reason.
{
enqueue(serializedJson);
return false;
return 0;
}
}
@ -528,7 +579,7 @@ namespace ix
{
auto webSocketSendInfo = _webSocket->send(serializedJson);
CobraConnection::invokeTrafficTrackerCallback(webSocketSendInfo.wireSize,
false);
false);
return webSocketSendInfo.success;
}

30
ws/ixcobra/IXCobraConnection.h
View File

@ -15,6 +15,7 @@
#include <string>
#include <thread>
#include <unordered_map>
#include <limits>
namespace ix
{
@ -27,7 +28,8 @@ namespace ix
CobraConnection_EventType_Open = 2,
CobraConnection_EventType_Closed = 3,
CobraConnection_EventType_Subscribed = 4,
CobraConnection_EventType_UnSubscribed = 5
CobraConnection_EventType_UnSubscribed = 5,
CobraConnection_EventType_Published = 6
};
enum CobraConnectionPublishMode
@ -40,12 +42,17 @@ namespace ix
using EventCallback = std::function<void(CobraConnectionEventType,
const std::string&,
const WebSocketHttpHeaders&,
const std::string&)>;
const std::string&,
uint64_t msgId)>;
using TrafficTrackerCallback = std::function<void(size_t size, bool incoming)>;
using PublishTrackerCallback = std::function<void(bool sent, bool acked)>;
class CobraConnection
{
public:
using MsgId = uint64_t;
CobraConnection();
~CobraConnection();
@ -57,11 +64,18 @@ namespace ix
const std::string& rolesecret,
const WebSocketPerMessageDeflateOptions& webSocketPerMessageDeflateOptions);
/// Set the traffic tracker callback
static void setTrafficTrackerCallback(const TrafficTrackerCallback& callback);
/// Reset the traffic tracker callback to an no-op one.
static void resetTrafficTrackerCallback();
/// Set the publish tracker callback
static void setPublishTrackerCallback(const PublishTrackerCallback& callback);
/// Reset the publish tracker callback to an no-op one.
static void resetPublishTrackerCallback();
/// Set the closed callback
void setEventCallback(const EventCallback& eventCallback);
@ -71,7 +85,7 @@ namespace ix
/// Publish a message to a channel
///
/// No-op if the connection is not established
bool publish(const Json::Value& channels, const Json::Value& msg);
MsgId publish(const Json::Value& channels, const Json::Value& msg);
// Subscribe to a channel, and execute a callback when an incoming
// message arrives.
@ -111,6 +125,7 @@ namespace ix
bool handleSubscriptionData(const Json::Value& pdu);
bool handleSubscriptionResponse(const Json::Value& pdu);
bool handleUnsubscriptionResponse(const Json::Value& pdu);
bool handlePublishResponse(const Json::Value& pdu);
void initWebSocketOnMessageCallback();
@ -121,11 +136,15 @@ namespace ix
/// Invoke the traffic tracker callback
static void invokeTrafficTrackerCallback(size_t size, bool incoming);
/// Invoke the publish tracker callback
static void invokePublishTrackerCallback(bool sent, bool acked);
/// Invoke event callbacks
void invokeEventCallback(CobraConnectionEventType eventType,
const std::string& errorMsg = std::string(),
const WebSocketHttpHeaders& headers = WebSocketHttpHeaders(),
const std::string& subscriptionId = std::string());
const std::string& subscriptionId = std::string(),
uint64_t msgId = std::numeric_limits<uint64_t>::max());
void invokeErrorCallback(const std::string& errorMsg, const std::string& serializedPdu);
///
@ -150,6 +169,9 @@ namespace ix
/// Traffic tracker callback
static TrafficTrackerCallback _trafficTrackerCallback;
/// Publish tracker callback
static PublishTrackerCallback _publishTrackerCallback;
/// Cobra events callbacks
EventCallback _eventCallback;
mutable std::mutex _eventCallbackMutex;

7
ws/ixcobra/IXCobraMetricsThreadedPublisher.cpp
View File

@ -25,7 +25,8 @@ namespace ix
(ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId)
const std::string& subscriptionId,
CobraConnection::MsgId msgId)
{
std::stringstream ss;
@ -58,6 +59,10 @@ namespace ix
{
ss << "Unsubscribed through subscription id: " << subscriptionId;
}
else if (eventType == ix::CobraConnection_EventType_Published)
{
ss << "Published message " << msgId << " acked";
}
ix::IXCoreLogger::Log(ss.str().c_str());
});