atria-soft/zeus
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

[DEV] access on the service with binary mode (if possible)

Browse Source
This commit is contained in:
Edouard DUPIN 2016-06-15 21:31:30 +02:00
parent 132f48b26f
commit 331c5f941b
11 changed files with 300 additions and 491 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
jus/Client.cpp
View File

@ -12,9 +12,8 @@
jus::Client::Client() : jus::Client::Client() :
propertyIp(this, "ip", "127.0.0.1", "Ip to connect server", &jus::Client::onPropertyChangeIp), propertyIp(this, "ip", "127.0.0.1", "Ip to connect server", &jus::Client::onPropertyChangeIp),
propertyPort(this, "port", 1983, "Port to connect server", &jus::Client::onPropertyChangePort), propertyPort(this, "port", 1983, "Port to connect server", &jus::Client::onPropertyChangePort) {
m_id(1) {
m_interfaceClient.connect(this, &jus::Client::onClientData);
} }
jus::Client::~Client() { jus::Client::~Client() {
@ -22,91 +21,11 @@ jus::Client::~Client() {
} }
void jus::Client::onClientData(jus::Buffer& _value) { void jus::Client::onClientData(jus::Buffer& _value) {
JUS_DEBUG("Get answer RAW : "/* << _value*/); JUS_ERROR("Get Data On the Communication interface that is not understand ... : " << _value.toJson().generateHumanString());
jus::FutureBase future;
uint64_t tid = _value.getTransactionId();
if (tid == 0) {
JUS_ERROR("Get a Protocol error ... No ID ...");
/*
if (obj["error"].toString().get() == "PROTOCOL-ERROR") {
JUS_ERROR("Get a Protocol error ...");
std::unique_lock<std::mutex> lock(m_mutex);
for (auto &it : m_pendingCall) {
if (it.isValid() == false) {
continue;
}
it.setAnswer(obj);
}
m_pendingCall.clear();
} else {
JUS_ERROR("call with no ID ==> error ...");
}
*/
return;
}
{
std::unique_lock<std::mutex> lock(m_mutex);
auto it = m_pendingCall.begin();
while (it != m_pendingCall.end()) {
if (it->isValid() == false) {
it = m_pendingCall.erase(it);
continue;
}
if (it->getTransactionId() != tid) {
++it;
continue;
}
future = *it;
break;
}
}
if (future.isValid() == false) {
JUS_TODO("manage this event better ...");
//m_newData.push_back(std::move(_value));
return;
}
bool ret = future.setAnswer(_value);
if (ret == true) {
std::unique_lock<std::mutex> lock(m_mutex);
auto it = m_pendingCall.begin();
while (it != m_pendingCall.end()) {
if (it->isValid() == false) {
it = m_pendingCall.erase(it);
continue;
}
if (it->getTransactionId() != tid) {
++it;
continue;
}
it = m_pendingCall.erase(it);
break;
}
}
} }
jus::ServiceRemote jus::Client::getService(const std::string& _name) { jus::ServiceRemote jus::Client::getService(const std::string& _name) {
return jus::ServiceRemote(this, _name); return jus::ServiceRemote(m_interfaceClient, _name);
}
int32_t jus::Client::link(const std::string& _serviceName) {
// TODO : Check the number of connection of this service ...
jus::Future<int32_t> ret = call("link", _serviceName);
ret.wait();
if (ret.hasError() == true) {
JUS_WARNING("Can not link with the service named: '" << _serviceName << "' ==> link error");
return false;
}
return ret.get();
}
bool jus::Client::unlink(const uint32_t& _serviceId) {
jus::Future<bool> ret = call("unlink", _serviceId);
ret.wait();
if (ret.hasError() == true) {
JUS_WARNING("Can not unlink with the service id: '" << _serviceId << "' ==> link error");
return false;
}
return ret.get();
} }
void jus::Client::onPropertyChangeIp() { void jus::Client::onPropertyChangeIp() {
@ -119,17 +38,23 @@ void jus::Client::onPropertyChangePort(){
bool jus::Client::connect(const std::string& _remoteUserToConnect){ bool jus::Client::connect(const std::string& _remoteUserToConnect){
disconnect();
JUS_DEBUG("connect [START]"); JUS_DEBUG("connect [START]");
disconnect();
enet::Tcp connection = std::move(enet::connectTcpClient(*propertyIp, *propertyPort)); enet::Tcp connection = std::move(enet::connectTcpClient(*propertyIp, *propertyPort));
m_interfaceClient.setInterface(std::move(connection)); m_interfaceClient = std::make_shared<jus::TcpString>();
m_interfaceClient.connect(); if (m_interfaceClient == nullptr) {
JUS_ERROR("Allocate connection error");
return false;
}
m_interfaceClient->connect(this, &jus::Client::onClientData);
m_interfaceClient->setInterface(std::move(connection));
m_interfaceClient->connect();
// Force mode binary: // Force mode binary:
JUS_WARNING("Request change in mode Binary"); JUS_WARNING("Request change in mode Binary");
jus::Future<bool> retBin = call("setMode", "BIN").wait(); jus::Future<bool> retBin = call("setMode", "BIN").wait();
if (retBin.get() == true) { if (retBin.get() == true) {
JUS_WARNING(" ==> accepted binary"); JUS_WARNING(" ==> accepted binary");
m_interfaceClient.setMode(jus::connectionMode::modeBinary); m_interfaceClient->setMode(jus::connectionMode::modeBinary);
JUS_INFO("Connection jump in BINARY ..."); JUS_INFO("Connection jump in BINARY ...");
} else { } else {
// stay in JSON // stay in JSON
@ -152,13 +77,11 @@ bool jus::Client::connect(const std::string& _remoteUserToConnect){
void jus::Client::disconnect() { void jus::Client::disconnect() {
JUS_DEBUG("disconnect [START]"); JUS_DEBUG("disconnect [START]");
m_interfaceClient.disconnect(); if (m_interfaceClient != nullptr) {
m_interfaceClient->disconnect();
m_interfaceClient.reset();
} else {
JUS_VERBOSE("Nothing to disconnect ...");
}
JUS_DEBUG("disconnect [STOP]"); JUS_DEBUG("disconnect [STOP]");
} }
uint64_t jus::Client::getId() {
return m_id++;
}

64
jus/Client.h
View File

@ -9,11 +9,11 @@
#include <eproperty/Value.h> #include <eproperty/Value.h>
#include <ejson/ejson.h> #include <ejson/ejson.h>
#include <jus/debug.h> #include <jus/debug.h>
#include <jus/ServiceRemote.h>
#include <chrono> #include <chrono>
#include <unistd.h> #include <unistd.h>
#include <jus/Future.h> #include <jus/Future.h>
#include <jus/connectionMode.h> #include <jus/connectionMode.h>
#include <jus/ServiceRemote.h>
namespace jus { namespace jus {
class Client : public eproperty::Interface { class Client : public eproperty::Interface {
@ -21,23 +21,34 @@ namespace jus {
public: public:
eproperty::Value<std::string> propertyIp; eproperty::Value<std::string> propertyIp;
eproperty::Value<uint16_t> propertyPort; eproperty::Value<uint16_t> propertyPort;
std::mutex m_mutex;
std::vector<jus::FutureBase> m_pendingCall;
public:
enum jus::connectionMode getMode() { return m_interfaceClient.getMode(); }
private: private:
jus::TcpString m_interfaceClient; ememory::SharedPtr<jus::TcpString> m_interfaceClient;
uint32_t m_id;
std::vector<std::string> m_newData;
public: public:
/**
* @brief Create a client on a specific user in a client mode with the tocken associated
* @param[in] _address Address of the user: "ABCD.efgh#atria-soft.com:1993"
* @param[in]
* @param[in]
*/
//Client(const std::string& _address, const std::string& _clientName, const std::string& _clientTocken);
/**
* @brief Create a client on a specific user in a user mode (connect to your personnal account)
* @param[in] _address Address of the user: "ABCD.efgh#atria-soft.com:1993"
* @param[in] _userPassword Password of the user
*/
//Client(const std::string& _address, const std::string& _userPassword);
/**
* @brief Create a client on a specific user in an ANONIMOUS way
* @param[in] _address Address of the user: "ABCD.efgh#atria-soft.com:1993"
*/
//Client(const std::string& _address);
Client(); Client();
virtual ~Client(); virtual ~Client();
bool connect(const std::string& _remoteUserToConnect); bool connect(const std::string& _remoteUserToConnect);
void disconnect(); void disconnect();
public: public:
jus::ServiceRemote getService(const std::string& _serviceName); jus::ServiceRemote getService(const std::string& _serviceName);
int32_t link(const std::string& _serviceName);
bool unlink(const uint32_t& _serviceId);
// Connect that is not us // Connect that is not us
//bool identify("clientTest1#atria-soft.com", "QSDQSDGQSF54HSXWVCSQDJ654URTDJ654NBXCDFDGAEZ51968"); //bool identify("clientTest1#atria-soft.com", "QSDQSDGQSF54HSXWVCSQDJ654URTDJ654NBXCDFDGAEZ51968");
@ -46,36 +57,23 @@ namespace jus {
private: private:
void onClientData(jus::Buffer& _value); void onClientData(jus::Buffer& _value);
public: public:
uint64_t getId();
template<class... _ARGS> template<class... _ARGS>
jus::FutureBase call(const std::string& _functionName, _ARGS&&... _args) { jus::FutureBase call(const std::string& _functionName, _ARGS&&... _args) {
return m_interfaceClient.call(_functionName, _args...); if (m_interfaceClient == nullptr) {
/* jus::Buffer ret;
uint64_t id = getId(); ret.addError("NULLPTR", "call " + _functionName + " with no interface open");
std::vector<ActionAsyncClient> asyncAction; return jus::FutureBase(0, true, ret);
if (getMode() == jus::connectionMode::modeJson) {
ejson::Object callElem = jus::createCall(asyncAction, id, _functionName, std::forward<_ARGS>(_args)...);
return callJson(id, callElem, asyncAction);
} else {
jus::Buffer callElem = jus::createBinaryCall(asyncAction, id, _functionName, std::forward<_ARGS>(_args)...);
return callBinary(id, callElem, asyncAction);
} }
*/ return m_interfaceClient->call(_functionName, _args...);
} }
template<class... _ARGS> template<class... _ARGS>
jus::FutureBase callAction(const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) { jus::FutureBase callAction(const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) {
return m_interfaceClient.callAction(_functionName, _args..., _callback); if (m_interfaceClient == nullptr) {
/* jus::Buffer ret;
uint64_t id = getId(); ret.addError("NULLPTR", "call " + _functionName + " with no interface open");
std::vector<ActionAsyncClient> asyncAction; return jus::FutureBase(0, true, ret, _callback);
if (getMode() == jus::connectionMode::modeJson) {
ejson::Object callElem = jus::createCall(asyncAction, id, _functionName, std::forward<_ARGS>(_args)...);
return callJson(id, callElem, asyncAction, _callback);
} else {
jus::Buffer callElem = jus::createBinaryCall(asyncAction, id, _functionName, std::forward<_ARGS>(_args)...);
return callBinary(id, callElem, asyncAction, _callback);
} }
*/ return m_interfaceClient->callAction(_functionName, _args..., _callback);
} }
private: private:
void onPropertyChangeIp(); void onPropertyChangeIp();

226
jus/GateWayClient.cpp
View File

@ -18,8 +18,7 @@ static const std::string protocolError = "PROTOCOL-ERROR";
jus::GateWayClient::GateWayClient(enet::Tcp _connection, jus::GateWay* _gatewayInterface) : jus::GateWayClient::GateWayClient(enet::Tcp _connection, jus::GateWay* _gatewayInterface) :
m_state(jus::GateWayClient::state::unconnect), m_state(jus::GateWayClient::state::unconnect),
m_gatewayInterface(_gatewayInterface), m_gatewayInterface(_gatewayInterface),
m_interfaceClient(std::move(_connection)), m_interfaceClient(std::move(_connection)) {
m_transactionLocalId(1) {
JUS_INFO("----------------"); JUS_INFO("----------------");
JUS_INFO("-- NEW Client --"); JUS_INFO("-- NEW Client --");
JUS_INFO("----------------"); JUS_INFO("----------------");
@ -47,18 +46,24 @@ void jus::GateWayClient::stop() {
if (it == nullptr) { if (it == nullptr) {
continue; continue;
} }
it->m_interfaceClient.callClient(m_uid, "_delete");
/*
ejson::Object linkService; ejson::Object linkService;
linkService.add("call", ejson::String("_delete")); linkService.add("call", ejson::String("_delete"));
linkService.add("id", ejson::Number(m_transactionLocalId++)); linkService.add("id", ejson::Number(m_transactionLocalId++));
linkService.add("param", ejson::Array()); linkService.add("param", ejson::Array());
it->SendData(m_uid, linkService); it->SendData(m_uid, linkService);
*/
} }
if (m_userService != nullptr) { if (m_userService != nullptr) {
m_userService->m_interfaceClient.callClient(m_uid2, "_delete");
/*
ejson::Object linkService; ejson::Object linkService;
linkService.add("call", ejson::String("_delete")); linkService.add("call", ejson::String("_delete"));
linkService.add("id", ejson::Number(m_transactionLocalId++)); linkService.add("id", ejson::Number(m_transactionLocalId++));
linkService.add("param", ejson::Array()); linkService.add("param", ejson::Array());
m_userService->SendData(m_uid2, linkService); m_userService->SendData(m_uid2, linkService);
*/
m_userService = nullptr; m_userService = nullptr;
} }
m_listConnectedService.clear(); m_listConnectedService.clear();
@ -131,7 +136,7 @@ void jus::GateWayClient::onClientData(jus::Buffer& _value) {
if (m_userService == nullptr) { if (m_userService == nullptr) {
answerProtocolError(transactionId, "Gateway internal error 'No user interface'"); answerProtocolError(transactionId, "Gateway internal error 'No user interface'");
} else { } else {
jus::Future<bool> futLocalService = call(m_uid2, m_userService, "_new", m_userConnectionName, "**Gateway**", std::vector<std::string>()); jus::Future<bool> futLocalService = m_userService->m_interfaceClient.callClient(m_uid2, "_new", m_userConnectionName, "**Gateway**", std::vector<std::string>());
futLocalService.wait(); // TODO: Set timeout ... futLocalService.wait(); // TODO: Set timeout ...
m_state = jus::GateWayClient::state::userIdentify; m_state = jus::GateWayClient::state::userIdentify;
m_interfaceClient.answerValue(transactionId, true); m_interfaceClient.answerValue(transactionId, true);
@ -161,7 +166,8 @@ void jus::GateWayClient::onClientData(jus::Buffer& _value) {
answerProtocolError(transactionId, "gateWay internal error 3"); answerProtocolError(transactionId, "gateWay internal error 3");
return; return;
} }
jus::Future<bool> fut = call(m_uid2, m_userService, "checkTocken", clientName, clientTocken);
jus::Future<bool> fut = m_userService->m_interfaceClient.callClient(m_uid2, "checkTocken", clientName, clientTocken);
fut.wait(); // TODO: Set timeout ... fut.wait(); // TODO: Set timeout ...
if (fut.hasError() == true) { if (fut.hasError() == true) {
JUS_ERROR("Get error from the service ..."); JUS_ERROR("Get error from the service ...");
@ -177,7 +183,7 @@ void jus::GateWayClient::onClientData(jus::Buffer& _value) {
} }
if (callFunction == "auth") { if (callFunction == "auth") {
std::string password = _value.getParameter<std::string>(0); std::string password = _value.getParameter<std::string>(0);
jus::Future<bool> fut = call(m_uid2, m_userService, "checkAuth", password); jus::Future<bool> fut = m_userService->m_interfaceClient.callClient(m_uid2, "checkAuth", password);
fut.wait(); // TODO: Set timeout ... fut.wait(); // TODO: Set timeout ...
if (fut.hasError() == true) { if (fut.hasError() == true) {
JUS_ERROR("Get error from the service ..."); JUS_ERROR("Get error from the service ...");
@ -197,7 +203,7 @@ void jus::GateWayClient::onClientData(jus::Buffer& _value) {
// -------------------------------- // --------------------------------
// -- Get groups: // -- Get groups:
// -------------------------------- // --------------------------------
jus::Future<std::vector<std::string>> futGroup = call(m_uid2, m_userService, "getGroups", m_clientName); jus::Future<std::vector<std::string>> futGroup = m_userService->m_interfaceClient.callClient(m_uid2, "getGroups", m_clientName);
futGroup.wait(); // TODO: Set timeout ... futGroup.wait(); // TODO: Set timeout ...
if (futGroup.hasError() == true) { if (futGroup.hasError() == true) {
JUS_ERROR("Get error from the service ..."); JUS_ERROR("Get error from the service ...");
@ -210,7 +216,7 @@ void jus::GateWayClient::onClientData(jus::Buffer& _value) {
// -- Get services: // -- Get services:
// -------------------------------- // --------------------------------
std::vector<std::string> currentServices = m_gatewayInterface->getAllServiceName(); std::vector<std::string> currentServices = m_gatewayInterface->getAllServiceName();
jus::Future<std::vector<std::string>> futServices = call(m_uid2, m_userService, "filterServices", m_clientName, currentServices); jus::Future<std::vector<std::string>> futServices = m_userService->m_interfaceClient.callClient(m_uid2, "filterServices", m_clientName, currentServices);
futServices.wait(); // TODO: Set timeout ... futServices.wait(); // TODO: Set timeout ...
if (futServices.hasError() == true) { if (futServices.hasError() == true) {
JUS_ERROR("Get error from the service ..."); JUS_ERROR("Get error from the service ...");
@ -267,7 +273,7 @@ void jus::GateWayClient::onClientData(jus::Buffer& _value) {
} }
ememory::SharedPtr<jus::GateWayService> srv = m_gatewayInterface->get(serviceName); ememory::SharedPtr<jus::GateWayService> srv = m_gatewayInterface->get(serviceName);
if (srv != nullptr) { if (srv != nullptr) {
jus::Future<bool> futLink = call(m_uid, srv, "_new", m_userConnectionName, m_clientName, m_clientgroups); jus::Future<bool> futLink = srv->m_interfaceClient.callClient(m_uid, "_new", m_userConnectionName, m_clientName, m_clientgroups);
futLink.wait(); // TODO: Set timeout ... futLink.wait(); // TODO: Set timeout ...
if (futLink.hasError() == true) { if (futLink.hasError() == true) {
JUS_ERROR("Get error from the service ... LINK"); JUS_ERROR("Get error from the service ... LINK");
@ -292,7 +298,7 @@ void jus::GateWayClient::onClientData(jus::Buffer& _value) {
m_interfaceClient.answerError(transactionId, "NOT-CONNECTED-SERVICE"); m_interfaceClient.answerError(transactionId, "NOT-CONNECTED-SERVICE");
return; return;
} }
jus::Future<bool> futUnLink = call(m_uid, m_listConnectedService[localServiceID], "_delete"); jus::Future<bool> futUnLink = m_listConnectedService[localServiceID]->m_interfaceClient.callClient(m_uid, "_delete");
futUnLink.wait(); // TODO: Set timeout ... futUnLink.wait(); // TODO: Set timeout ...
if (futUnLink.hasError() == true) { if (futUnLink.hasError() == true) {
JUS_ERROR("Get error from the service ... UNLINK"); JUS_ERROR("Get error from the service ... UNLINK");
@ -318,185 +324,43 @@ void jus::GateWayClient::onClientData(jus::Buffer& _value) {
JUS_ERROR("TODO : Manage this case ..."); JUS_ERROR("TODO : Manage this case ...");
return; return;
} }
bool finish = _value.getPartFinish();
uint16_t partId = _value.getPartId(); uint16_t partId = _value.getPartId();
if (partId != 0) { if (partId != 0) {
// subMessage ... ==> try to forward message: m_listConnectedService[serviceId]->m_interfaceClient.callForwardMultiple(
std::unique_lock<std::mutex> lock(m_mutex); m_uid,
for (auto &itCall : m_pendingCall) { _value,
JUS_INFO(" compare : " << itCall.first << " =?= " << transactionId); (uint64_t(m_uid) << 32) + uint64_t(transactionId));
if (itCall.first == transactionId) {
// Find element ==> transit it ...
_value.setTransactionId(itCall.second.getTransactionId());
m_listConnectedService[serviceId]->SendData(m_uid, _value);
return;
}
}
JUS_ERROR("Can not transfer part of a message ...");
return; return;
} }
callActionForward(m_listConnectedService[serviceId], m_listConnectedService[serviceId]->m_interfaceClient.callForward(
_value, m_uid,
[=](jus::FutureBase _ret) { _value,
(uint64_t(m_uid) << 32) + uint64_t(transactionId),
// TODO : Check if it is a JSON or binary ... [=](jus::FutureBase _ret) {
jus::Buffer tmpp = _ret.getRaw(); jus::Buffer tmpp = _ret.getRaw();
JUS_DEBUG(" ==> transmit : " << tmpp.getTransactionId() << " -> " << transactionId); JUS_DEBUG(" ==> transmit : " << tmpp.getTransactionId() << " -> " << transactionId);
JUS_DEBUG(" msg=" << tmpp.generateHumanString()); JUS_DEBUG(" msg=" << tmpp.generateHumanString());
tmpp.setTransactionId(transactionId); tmpp.setTransactionId(transactionId);
JUS_DEBUG("transmit=" << tmpp.generateHumanString()); tmpp.setServiceId(serviceId+1);
m_interfaceClient.writeBinary(tmpp); JUS_DEBUG("transmit=" << tmpp.generateHumanString());
// multiple send element ... if (m_interfaceClient.getMode() == jus::connectionMode::modeJson) {
return tmpp.getPartFinish(); ejson::Object obj = tmpp.toJson();
/* m_interfaceClient.writeJson(obj);
// TODO : Check if it is a JSON or binary ... } else if (m_interfaceClient.getMode() == jus::connectionMode::modeBinary) {
ejson::Object tmpp = _ret.getRaw(); m_interfaceClient.writeBinary(tmpp);
JUS_VERBOSE(" ==> transmit : " << tmpp["id"].toNumber().getU64() << " -> " << transactionId); } else if (m_interfaceClient.getMode() == jus::connectionMode::modeXml) {
JUS_VERBOSE(" msg=" << tmpp.generateMachineString()); JUS_ERROR("TODO ... ");
tmpp["id"].toNumber().set(uint64_t(transactionId)); } else {
m_interfaceClient.writeJson(tmpp); JUS_ERROR("wrong type of communication");
if (tmpp.valueExist("part") == true) { }
// multiple send element ... // multiple send element ...
if (tmpp.valueExist("finish") == true) { return tmpp.getPartFinish();
return tmpp["finish"].toBoolean().get(); });
}
return false;
}
return true;
*/
});
} }
} }
} }
} }
jus::FutureBase jus::GateWayClient::callBinary(uint64_t _callerId,
ememory::SharedPtr<jus::GateWayService> _srv,
uint64_t _clientTransactionId,
uint64_t _transactionId,
jus::Buffer& _obj,
jus::FutureData::ObserverFinish _callback) {
JUS_VERBOSE("Send BINARY [START] ");
if (_srv == nullptr) {
// TODO : Change this ...
jus::Buffer obj;
obj.setTransactionId(_transactionId);
obj.setClientId(_callerId);
obj.setType(jus::Buffer::typeMessage::answer);
obj.addError("NOT-CONNECTED", "Client interface not connected (no TCP)");
return jus::FutureBase(_transactionId, true, obj, _callback);
}
jus::FutureBase tmpFuture(_transactionId, _callback);
{
std::unique_lock<std::mutex> lock(m_mutex);
m_pendingCall.push_back(std::make_pair(_clientTransactionId, tmpFuture));
}
_obj.setTransactionId(_transactionId);
_srv->SendData(_callerId, _obj);
JUS_VERBOSE("Send BINARY [STOP]");
return tmpFuture;
}
jus::FutureBase jus::GateWayClient::callActionForward(ememory::SharedPtr<jus::GateWayService> _srv,
jus::Buffer& _Buffer,
jus::FutureData::ObserverFinish _callback) {
uint64_t id = getId();
uint64_t clientTransactionId = _Buffer.getTransactionId();
jus::FutureBase ret = callBinary(m_uid, _srv, clientTransactionId, id, _Buffer, _callback);
ret.setSynchronous();
return ret;
}
uint64_t jus::GateWayClient::getId() {
return m_transactionLocalId++;
}
jus::FutureBase jus::GateWayClient::callJson(uint64_t _callerId,
ememory::SharedPtr<jus::GateWayService> _srv,
uint64_t _clientTransactionId,
uint64_t _transactionId,
const ejson::Object& _obj,
jus::FutureData::ObserverFinish _callback) {
JUS_VERBOSE("Send JSON [START] ");
if (_srv == nullptr) {
ejson::Object obj;
obj.add("error", ejson::String("NOT-CONNECTED"));
obj.add("error-help", ejson::String("Client interface not connected (no TCP)"));
return jus::FutureBase(_transactionId, true, obj, _callback);
}
jus::FutureBase tmpFuture(_transactionId, _callback);
{
std::unique_lock<std::mutex> lock(m_mutex);
m_pendingCall.push_back(std::make_pair(_clientTransactionId, tmpFuture));
}
_srv->SendData(_callerId, _obj);
JUS_VERBOSE("Send JSON [STOP]");
return tmpFuture;
}
void jus::GateWayClient::returnMessage(jus::Buffer& _data) { void jus::GateWayClient::returnMessage(jus::Buffer& _data) {
jus::FutureBase future; JUS_ERROR("Get call from the Service to the user ...");
uint32_t tid = _data.getTransactionId(); }
if (tid == 0) {
/* TODO ...
if (_data["error"].toString().get() == "PROTOCOL-ERROR") {
JUS_ERROR("Get a Protocol error ...");
std::unique_lock<std::mutex> lock(m_mutex);
for (auto &it : m_pendingCall) {
if (it.second.isValid() == false) {
continue;
}
it.second.setAnswer(_data);
}
m_pendingCall.clear();
} else {
JUS_ERROR("call with no ID ==> error ...");
}
*/
return;
}
{
std::unique_lock<std::mutex> lock(m_mutex);
auto it = m_pendingCall.begin();
while (it != m_pendingCall.end()) {
if (it->second.isValid() == false) {
it = m_pendingCall.erase(it);
continue;
}
if (it->second.getTransactionId() != tid) {
++it;
continue;
}
// TODO : Do it better ...
future = it->second;
break;
}
}
if (future.isValid() == false) {
JUS_WARNING("Action to do ...");
return;
}
bool ret = future.setAnswer(_data);
if (ret == true) {
std::unique_lock<std::mutex> lock(m_mutex);
auto it = m_pendingCall.begin();
while (it != m_pendingCall.end()) {
if (it->second.isValid() == false) {
it = m_pendingCall.erase(it);
continue;
}
if (it->second.getTransactionId() != tid) {
++it;
continue;
}
it = m_pendingCall.erase(it);
break;
}
}
}

54
jus/GateWayClient.h
View File

@ -57,60 +57,6 @@ namespace jus {
} }
bool isAlive(); bool isAlive();
private:
std::mutex m_mutex;
std::vector<std::pair<uint64_t, jus::FutureBase>> m_pendingCall;
int32_t m_transactionLocalId;
jus::FutureBase callJson(uint64_t _callerId,
ememory::SharedPtr<jus::GateWayService> _srv,
uint64_t _clientTransactionId,
uint64_t _transactionId,
const ejson::Object& _obj,
jus::FutureData::ObserverFinish _callback=nullptr);
jus::FutureBase callBinary(uint64_t _callerId,
ememory::SharedPtr<jus::GateWayService> _srv,
uint64_t _clientTransactionId,
uint64_t _transactionId,
jus::Buffer& _obj,
jus::FutureData::ObserverFinish _callback=nullptr);
uint64_t getId();
public:
template<class... _ARGS>
jus::FutureBase call(uint64_t _callerId, ememory::SharedPtr<jus::GateWayService> _srv, const std::string& _functionName, _ARGS&&... _args) {
uint64_t id = getId();
std::vector<ActionAsyncClient> asyncAction;
ejson::Object callElem = jus::createCall(asyncAction, id, _functionName, std::forward<_ARGS>(_args)...);
if (asyncAction.size() != 0) {
JUS_ERROR("Missing send async messages");
}
return callJson(_callerId, _srv, 0, id, callElem);
}
template<class... _ARGS>
jus::FutureBase callAction(uint64_t _callerId, ememory::SharedPtr<jus::GateWayService> _srv, const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) {
uint64_t id = getId();
std::vector<ActionAsyncClient> asyncAction;
ejson::Object callElem = jus::createCall(asyncAction, id, _functionName, std::forward<_ARGS>(_args)...);
if (asyncAction.size() != 0) {
JUS_ERROR("Missing send async messages");
}
return callJson(_callerId, _srv, 0, id, callElem, _callback);
}
jus::FutureBase callActionForward(uint64_t _callerId,
uint64_t _clientTransactionId,
ememory::SharedPtr<jus::GateWayService> _srv,
const std::string& _functionName,
ejson::Array _params,
jus::FutureData::ObserverFinish _callback,
int64_t _part,
bool _finish);
jus::FutureBase callActionForward(ememory::SharedPtr<jus::GateWayService> _srv,
jus::Buffer& _Buffer,
jus::FutureData::ObserverFinish _callback);
void answerProtocolError(uint32_t _transactionId, const std::string& _errorHelp); void answerProtocolError(uint32_t _transactionId, const std::string& _errorHelp);

4
jus/GateWayService.cpp
View File

@ -64,7 +64,7 @@ void jus::GateWayService::SendData(uint64_t _userSessionId, jus::Buffer& _data)
} }
void jus::GateWayService::onServiceData(jus::Buffer& _value) { void jus::GateWayService::onServiceData(jus::Buffer& _value) {
//JUS_DEBUG("On service data: " << _value); JUS_DEBUG("On service data: " << _value.toJson().generateHumanString());
uint32_t transactionId = _value.getTransactionId(); uint32_t transactionId = _value.getTransactionId();
//data.add("from-service", ejson::String(m_name)); //data.add("from-service", ejson::String(m_name));
if (_value.getType() == jus::Buffer::typeMessage::event) { if (_value.getType() == jus::Buffer::typeMessage::event) {
@ -114,7 +114,7 @@ void jus::GateWayService::onServiceData(jus::Buffer& _value) {
} }
m_name = _value.getParameter<std::string>(0); m_name = _value.getParameter<std::string>(0);
m_interfaceClient.setInterfaceName("srv-" + m_name); m_interfaceClient.setInterfaceName("srv-" + m_name);
m_interfaceClient.answerValue(transactionId, false); m_interfaceClient.answerValue(transactionId, true);
return; return;
} }
answerProtocolError(transactionId, "unknow function"); answerProtocolError(transactionId, "unknow function");

2
jus/GateWayService.h
View File

@ -13,7 +13,9 @@
namespace jus { namespace jus {
class GateWay; class GateWay;
class GateWayClient;
class GateWayService { class GateWayService {
friend class jus::GateWayClient;
private: private:
jus::GateWay* m_gatewayInterface; jus::GateWay* m_gatewayInterface;
jus::TcpString m_interfaceClient; jus::TcpString m_interfaceClient;

85
jus/Service.cpp
View File

@ -15,8 +15,7 @@
jus::Service::Service() : jus::Service::Service() :
propertyIp(this, "ip", "127.0.0.1", "Ip to connect server", &jus::Service::onPropertyChangeIp), propertyIp(this, "ip", "127.0.0.1", "Ip to connect server", &jus::Service::onPropertyChangeIp),
propertyPort(this, "port", 1982, "Port to connect server", &jus::Service::onPropertyChangePort) { propertyPort(this, "port", 1982, "Port to connect server", &jus::Service::onPropertyChangePort) {
m_interfaceClient = std::make_shared<jus::TcpString>();
m_interfaceClient->connect(this, &jus::Service::onClientData);
advertise("getExtention", &jus::Service::getExtention); advertise("getExtention", &jus::Service::getExtention);
setLastFuncDesc("Get List of availlable extention of this service"); setLastFuncDesc("Get List of availlable extention of this service");
@ -54,6 +53,7 @@ void jus::Service::onClientData(jus::Buffer& _value) {
} }
jus::FutureCall futCall(clientId, tmpID, _value); jus::FutureCall futCall(clientId, tmpID, _value);
if (futCall.isFinished() == true) { if (futCall.isFinished() == true) {
JUS_INFO("Call Binary ..");
callBinary(tmpID, futCall.getRaw()); callBinary(tmpID, futCall.getRaw());
} else { } else {
m_callMultiData.push_back(futCall); m_callMultiData.push_back(futCall);
@ -77,25 +77,48 @@ void jus::Service::connect(const std::string& _serviceName, uint32_t _numberRetr
JUS_DEBUG("connect [STOP] ==> can not connect"); JUS_DEBUG("connect [STOP] ==> can not connect");
return; return;
} }
m_interfaceClient = std::make_shared<jus::TcpString>();
if (m_interfaceClient == nullptr) {
JUS_ERROR("Can not allocate interface ...");
return;
}
m_interfaceClient->connect(this, &jus::Service::onClientData);
m_interfaceClient->setInterface(std::move(connection)); m_interfaceClient->setInterface(std::move(connection));
m_interfaceClient->connect(); m_interfaceClient->connect();
ejson::Object tmpp; jus::Future<bool> ret = m_interfaceClient->call("setMode", "BIN");
tmpp.add("id", ejson::Number(1)); ret.wait();
tmpp.add("call", ejson::String("connect-service")); if (ret.get() == false) {
ejson::Array params; JUS_ERROR("Can not communicate with the gateway in Binary mode ... STAY in JSON");
params.add(ejson::String(_serviceName)); } else {
tmpp.add("param", params); JUS_INFO("Change mode in Binary");
m_interfaceClient->writeJson(tmpp); m_interfaceClient->setMode(jus::connectionMode::modeBinary);
}
ret = m_interfaceClient->call("connect-service", _serviceName);
ret.wait();
if (ret.get() == false) {
JUS_ERROR("Can not configure the interface for the service with the current name ...");
m_interfaceClient->disconnect();
return;
}
JUS_DEBUG("connect [STOP]"); JUS_DEBUG("connect [STOP]");
} }
void jus::Service::disconnect(){ void jus::Service::disconnect(){
JUS_DEBUG("disconnect [START]"); JUS_DEBUG("disconnect [START]");
m_interfaceClient->disconnect(); if (m_interfaceClient != nullptr) {
m_interfaceClient->disconnect();
m_interfaceClient.reset();
} else {
JUS_VERBOSE("Nothing to disconnect ...");
}
JUS_DEBUG("disconnect [STOP]"); JUS_DEBUG("disconnect [STOP]");
} }
bool jus::Service::GateWayAlive() { bool jus::Service::GateWayAlive() {
if (m_interfaceClient == nullptr) {
return false;
}
return m_interfaceClient->isActive(); return m_interfaceClient->isActive();
} }
@ -128,30 +151,26 @@ void jus::Service::callBinary(uint32_t _transactionId, jus::Buffer& _obj) {
return; return;
} }
//if (_obj.getType() == jus::Buffer::typeMessage::event) { //if (_obj.getType() == jus::Buffer::typeMessage::event) {
if (m_interfaceClient->getMode() == jus::connectionMode::modeBinary) { uint32_t clientId = _obj.getClientId();
std::string callFunction = _obj.getCall();
} else if (m_interfaceClient->getMode() == jus::connectionMode::modeJson) { if (callFunction[0] == '_') {
uint32_t clientId = _obj.getClientId(); if (callFunction == "_new") {
std::string callFunction = _obj.getCall(); std::string userName = _obj.getParameter<std::string>(0);
if (callFunction[0] == '_') { std::string clientName = _obj.getParameter<std::string>(1);
if (callFunction == "_new") { std::vector<std::string> clientGroup = _obj.getParameter<std::vector<std::string>>(2);
std::string userName = _obj.getParameter<std::string>(0); clientConnect(clientId, userName, clientName, clientGroup);
std::string clientName = _obj.getParameter<std::string>(1); } else if (callFunction == "_delete") {
std::vector<std::string> clientGroup = _obj.getParameter<std::vector<std::string>>(2); clientDisconnect(clientId);
clientConnect(clientId, userName, clientName, clientGroup);
} else if (callFunction == "_delete") {
clientDisconnect(clientId);
}
m_interfaceClient->answerValue(_transactionId, true, clientId);
return;
} else if (isFunctionAuthorized(clientId, callFunction) == true) {
callBinary2(_transactionId, clientId, callFunction, _obj);
return;
} else {
m_interfaceClient->answerError(_transactionId, "NOT-AUTHORIZED-FUNCTION", "", clientId);
return;
} }
m_interfaceClient->answerValue(_transactionId, true, clientId);
return;
} else if (isFunctionAuthorized(clientId, callFunction) == true) {
callBinary2(_transactionId, clientId, callFunction, _obj);
return;
} else { } else {
JUS_ERROR("Not manage transfer mode "); m_interfaceClient->answerError(_transactionId, "NOT-AUTHORIZED-FUNCTION", "", clientId);
return;
} }
} }

66
jus/ServiceRemote.cpp
View File

@ -7,24 +7,43 @@
#include <jus/ServiceRemote.h> #include <jus/ServiceRemote.h>
#include <jus/Client.h> #include <jus/Client.h>
jus::ServiceRemote::ServiceRemote(jus::Client* _clientInterface, const std::string& _name): jus::ServiceRemote::ServiceRemote(ememory::SharedPtr<jus::TcpString> _clientLink, const std::string& _name):
m_clientInterface(_clientInterface), m_interfaceClient(_clientLink),
m_name(_name), m_name(_name),
m_serviceId(0) { m_serviceId(0),
int32_t val = m_clientInterface->link(_name); m_isLinked(false) {
if (val >= 0) { if (m_interfaceClient == nullptr) {
m_isLinked = true; return;
m_serviceId = val;
} else {
m_isLinked = false;
m_serviceId = 0;
} }
// little hack : Call the service manager with the service ID=0 ...
jus::Future<uint32_t> ret = call("link", _name);
ret.wait();
if (ret.hasError() == true) {
JUS_WARNING("Can not link with the service named: '" << _name << "' ==> link error");
return;
}
m_isLinked = true;
m_serviceId = ret.get();
} }
jus::ServiceRemote::~ServiceRemote() { jus::ServiceRemote::~ServiceRemote() {
if (m_isLinked == true) { if (m_isLinked == true) {
m_clientInterface->unlink(m_serviceId); uint32_t tmpLocalService = m_serviceId;
m_isLinked = false; // little hack : Call the service manager with the service ID=0 ...
m_serviceId = 0;
jus::Future<bool> ret = call("unlink", tmpLocalService);
ret.wait();
if (ret.hasError() == true) {
JUS_WARNING("Can not unlink with the service id: '" << tmpLocalService << "' ==> link error");
m_serviceId = tmpLocalService;
return;
}
if (ret.get() == true) {
m_isLinked = false;
} else {
JUS_ERROR("Can not unlink with this service ....");
m_serviceId = tmpLocalService;
}
} }
} }
@ -32,26 +51,3 @@ bool jus::ServiceRemote::exist() {
return m_isLinked; return m_isLinked;
} }
uint64_t jus::ServiceRemote::getId() {
return m_clientInterface->getId();
}
enum jus::connectionMode jus::ServiceRemote::getMode() {
return m_clientInterface->getMode();
}
jus::FutureBase jus::ServiceRemote::callJson(uint64_t _transactionId,
const ejson::Object& _obj,
const std::vector<ActionAsyncClient>& _async,
jus::FutureData::ObserverFinish _callback) {
return m_clientInterface->callJson(_transactionId, _obj, _async, _callback, m_serviceId);
}
jus::FutureBase jus::ServiceRemote::callBinary(uint64_t _transactionId,
jus::Buffer& _obj,
const std::vector<ActionAsyncClient>& _async,
jus::FutureData::ObserverFinish _callback) {
return m_clientInterface->callBinary(_transactionId, _obj, _async, _callback, m_serviceId);
}

42
jus/ServiceRemote.h
View File

@ -13,54 +13,38 @@
#include <jus/ServiceRemote.h> #include <jus/ServiceRemote.h>
#include <jus/Future.h> #include <jus/Future.h>
#include <jus/connectionMode.h> #include <jus/connectionMode.h>
#include <jus/TcpString.h>
namespace jus { namespace jus {
class Client; class Client;
class ServiceRemote { class ServiceRemote {
private: private:
jus::Client* m_clientInterface; ememory::SharedPtr<jus::TcpString> m_interfaceClient;
std::string m_name; std::string m_name;
uint32_t m_serviceId; uint32_t m_serviceId;
bool m_isLinked; bool m_isLinked;
public: public:
ServiceRemote(jus::Client* _clientInterface, const std::string& _name); ServiceRemote(ememory::SharedPtr<jus::TcpString> _clientLink, const std::string& _name);
~ServiceRemote(); ~ServiceRemote();
bool exist(); bool exist();
private:
jus::FutureBase callJson(uint64_t _transactionId, const ejson::Object& _obj, const std::vector<ActionAsyncClient>& _async, jus::FutureData::ObserverFinish _callback=nullptr);
jus::FutureBase callBinary(uint64_t _transactionId, jus::Buffer& _obj, const std::vector<ActionAsyncClient>& _async, jus::FutureData::ObserverFinish _callback=nullptr);
uint64_t getId();
enum jus::connectionMode getMode();
public: public:
template<class... _ARGS> template<class... _ARGS>
jus::FutureBase call(const std::string& _functionName, _ARGS&&... _args) { jus::FutureBase call(const std::string& _functionName, _ARGS&&... _args) {
return m_clientInterface->m_interfaceClient.callService(m_serviceId, _functionName, _args...); if (m_interfaceClient == nullptr) {
/* jus::Buffer ret;
uint64_t id = getId(); ret.addError("NULLPTR", "call " + _functionName + " with no interface open");
std::vector<ActionAsyncClient> asyncActionToDo; return jus::FutureBase(0, true, ret);
if (getMode() == jus::connectionMode::modeJson) {
ejson::Object callElem = jus::createCallService(asyncActionToDo, id, m_serviceId, _functionName, std::forward<_ARGS>(_args)...);
return callJson(id, callElem, asyncActionToDo);
} else {
jus::Buffer callElem = jus::createBinaryCallService(asyncActionToDo, id, m_serviceId, _functionName, std::forward<_ARGS>(_args)...);
return callBinary(id, callElem, asyncActionToDo);
} }
*/ return m_interfaceClient->callService(m_serviceId, _functionName, _args...);
} }
template<class... _ARGS> template<class... _ARGS>
jus::FutureBase callAction(const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) { jus::FutureBase callAction(const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) {
return m_clientInterface->m_interfaceClient.callServiceAction(m_serviceId, _functionName, _args..., _callback); if (m_interfaceClient == nullptr) {
/* jus::Buffer ret;
uint64_t id = getId(); ret.addError("NULLPTR", "call " + _functionName + " with no interface open");
std::vector<ActionAsyncClient> asyncActionToDo; return jus::FutureBase(0, true, ret, _callback);
if (getMode() == jus::connectionMode::modeJson) {
ejson::Object callElem = jus::createCallService(asyncActionToDo, id, m_serviceId, _functionName, std::forward<_ARGS>(_args)...);
return callJson(id, callElem, asyncActionToDo, _callback);
} else {
jus::Buffer callElem = jus::createBinaryCallService(asyncActionToDo, id, m_serviceId, _functionName, std::forward<_ARGS>(_args)...);
return callBinary(id, callElem, asyncActionToDo, _callback);
} }
*/ return m_interfaceClient->callServiceAction(m_serviceId, _functionName, _args..., _callback);
} }
}; };
} }

92
jus/TcpString.cpp
View File

@ -16,7 +16,7 @@ jus::TcpString::TcpString(enet::Tcp _connection) :
m_threadRunning = false; m_threadRunning = false;
m_threadAsyncRunning = false; m_threadAsyncRunning = false;
m_interfaceMode = jus::connectionMode::modeJson; m_interfaceMode = jus::connectionMode::modeJson;
m_transmissionId = 0; m_transmissionId = 1;
} }
jus::TcpString::TcpString() : jus::TcpString::TcpString() :
@ -27,7 +27,7 @@ jus::TcpString::TcpString() :
m_threadRunning = false; m_threadRunning = false;
m_threadAsyncRunning = false; m_threadAsyncRunning = false;
m_interfaceMode = jus::connectionMode::modeJson; m_interfaceMode = jus::connectionMode::modeJson;
m_transmissionId = 0; m_transmissionId = 1;
} }
void jus::TcpString::setInterface(enet::Tcp _connection) { void jus::TcpString::setInterface(enet::Tcp _connection) {
@ -259,19 +259,20 @@ void jus::TcpString::newBuffer(jus::Buffer& _buffer) {
std::unique_lock<std::mutex> lock(m_pendingCallMutex); std::unique_lock<std::mutex> lock(m_pendingCallMutex);
auto it = m_pendingCall.begin(); auto it = m_pendingCall.begin();
while (it != m_pendingCall.end()) { while (it != m_pendingCall.end()) {
if (it->isValid() == false) { if (it->second.isValid() == false) {
it = m_pendingCall.erase(it); it = m_pendingCall.erase(it);
continue; continue;
} }
if (it->getTransactionId() != tid) { if (it->second.getTransactionId() != tid) {
++it; ++it;
continue; continue;
} }
future = *it; future = it->second;
break; break;
} }
} }
if (future.isValid() == false) { if (future.isValid() == false) {
// not a pending call ==> simple event or call ...
if (m_observerElement != nullptr) { if (m_observerElement != nullptr) {
m_observerElement(_buffer); m_observerElement(_buffer);
} }
@ -282,11 +283,11 @@ void jus::TcpString::newBuffer(jus::Buffer& _buffer) {
std::unique_lock<std::mutex> lock(m_pendingCallMutex); std::unique_lock<std::mutex> lock(m_pendingCallMutex);
auto it = m_pendingCall.begin(); auto it = m_pendingCall.begin();
while (it != m_pendingCall.end()) { while (it != m_pendingCall.end()) {
if (it->isValid() == false) { if (it->second.isValid() == false) {
it = m_pendingCall.erase(it); it = m_pendingCall.erase(it);
continue; continue;
} }
if (it->getTransactionId() != tid) { if (it->second.getTransactionId() != tid) {
++it; ++it;
continue; continue;
} }
@ -371,16 +372,16 @@ jus::FutureBase jus::TcpString::callJson(uint64_t _transactionId,
jus::FutureData::ObserverFinish _callback, jus::FutureData::ObserverFinish _callback,
const uint32_t& _serviceId) { const uint32_t& _serviceId) {
JUS_VERBOSE("Send JSON [START] "); JUS_VERBOSE("Send JSON [START] ");
if (m_interfaceClient.isActive() == false) { if (isActive() == false) {
jus::Buffer obj; jus::Buffer obj;
obj.setType(jus::Buffer::typeMessage::answer) { obj.setType(jus::Buffer::typeMessage::answer);
obj.addError("NOT-CONNECTED", "Client interface not connected (no TCP)"); obj.addError("NOT-CONNECTED", "Client interface not connected (no TCP)");
return jus::FutureBase(_transactionId, true, obj, _callback); return jus::FutureBase(_transactionId, true, obj, _callback);
} }
jus::FutureBase tmpFuture(_transactionId, _callback); jus::FutureBase tmpFuture(_transactionId, _callback);
{ {
std::unique_lock<std::mutex> lock(m_pendingCallMutex); std::unique_lock<std::mutex> lock(m_pendingCallMutex);
m_pendingCall.push_back(tmpFuture); m_pendingCall.push_back(std::make_pair(uint64_t(0), tmpFuture));
} }
if (_async.size() != 0) { if (_async.size() != 0) {
_obj.add("part", ejson::Number(0)); _obj.add("part", ejson::Number(0));
@ -440,16 +441,16 @@ jus::FutureBase jus::TcpString::callBinary(uint64_t _transactionId,
jus::FutureData::ObserverFinish _callback, jus::FutureData::ObserverFinish _callback,
const uint32_t& _serviceId) { const uint32_t& _serviceId) {
JUS_VERBOSE("Send Binary [START] "); JUS_VERBOSE("Send Binary [START] ");
if (m_interfaceClient.isActive() == false) { if (isActive() == false) {
jus::Buffer obj; jus::Buffer obj;
obj.setType(jus::Buffer::typeMessage::answer) { obj.setType(jus::Buffer::typeMessage::answer);
obj.addError("NOT-CONNECTED", "Client interface not connected (no TCP)"); obj.addError("NOT-CONNECTED", "Client interface not connected (no TCP)");
return jus::FutureBase(_transactionId, true, obj, _callback); return jus::FutureBase(_transactionId, true, obj, _callback);
} }
jus::FutureBase tmpFuture(_transactionId, _callback); jus::FutureBase tmpFuture(_transactionId, _callback);
{ {
std::unique_lock<std::mutex> lock(m_pendingCallMutex); std::unique_lock<std::mutex> lock(m_pendingCallMutex);
m_pendingCall.push_back(tmpFuture); m_pendingCall.push_back(std::make_pair(uint64_t(0), tmpFuture));
} }
if (_async.size() != 0) { if (_async.size() != 0) {
_obj.setPartFinish(false); _obj.setPartFinish(false);
@ -465,10 +466,69 @@ jus::FutureBase jus::TcpString::callBinary(uint64_t _transactionId,
return tmpFuture; return tmpFuture;
} }
jus::FutureBase jus::TcpString::callForward(uint32_t _clientId,
jus::Buffer& _buffer,
uint64_t _singleReferenceId,
jus::FutureData::ObserverFinish _callback) {
JUS_VERBOSE("Call Forward [START]");
//jus::FutureBase ret = callBinary(id, _Buffer, async, _callback);
//ret.setSynchronous();
if (isActive() == false) {
jus::Buffer obj;
obj.setType(jus::Buffer::typeMessage::answer);
obj.addError("NOT-CONNECTED", "Client interface not connected (no TCP)");
return jus::FutureBase(0, true, obj, _callback);
}
uint64_t id = getId();
_buffer.setTransactionId(id);
_buffer.setClientId(_clientId);
jus::FutureBase tmpFuture(id, _callback);
tmpFuture.setSynchronous();
{
std::unique_lock<std::mutex> lock(m_pendingCallMutex);
m_pendingCall.push_back(std::make_pair(_singleReferenceId, tmpFuture));
}
if (m_interfaceMode == jus::connectionMode::modeJson) {
ejson::Object obj = _buffer.toJson();
writeJson(obj);
} else if (m_interfaceMode == jus::connectionMode::modeBinary) {
writeBinary(_buffer);
} else if (m_interfaceMode == jus::connectionMode::modeXml) {
JUS_ERROR("TODO ... ");
} else {
JUS_ERROR("wrong type of communication");
}
JUS_VERBOSE("Send Forward [STOP]");
return tmpFuture;
}
void jus::TcpString::callForwardMultiple(uint32_t _clientId,
jus::Buffer& _buffer,
uint64_t _singleReferenceId){
// subMessage ... ==> try to forward message:
std::unique_lock<std::mutex> lock(m_pendingCallMutex);
for (auto &itCall : m_pendingCall) {
JUS_INFO(" compare : " << itCall.first << " =?= " << _singleReferenceId);
if (itCall.first == _singleReferenceId) {
// Find element ==> transit it ...
_buffer.setTransactionId(itCall.second.getTransactionId());
_buffer.setClientId(_clientId);
if (m_interfaceMode == jus::connectionMode::modeJson) {
ejson::Object obj = _buffer.toJson();
writeJson(obj);
} else if (m_interfaceMode == jus::connectionMode::modeBinary) {
writeBinary(_buffer);
} else if (m_interfaceMode == jus::connectionMode::modeXml) {
JUS_ERROR("TODO ... ");
} else {
JUS_ERROR("wrong type of communication");
}
return;
}
}
JUS_ERROR("Can not transfer part of a message ...");
}
void jus::TcpString::answerError(uint64_t _clientTransactionId, const std::string& _errorValue, const std::string& _errorHelp, uint32_t _clientId) { void jus::TcpString::answerError(uint64_t _clientTransactionId, const std::string& _errorValue, const std::string& _errorHelp, uint32_t _clientId) {
if (m_interfaceMode == jus::connectionMode::modeJson) { if (m_interfaceMode == jus::connectionMode::modeJson) {

39
jus/TcpString.h
View File

@ -12,6 +12,7 @@
#include <memory> #include <memory>
#include <jus/connectionMode.h> #include <jus/connectionMode.h>
#include <jus/AbstractFunction.h> #include <jus/AbstractFunction.h>
#include <jus/FutureBase.h>
namespace jus { namespace jus {
class TcpString : public eproperty::Interface { class TcpString : public eproperty::Interface {
@ -24,7 +25,7 @@ namespace jus {
return m_transmissionId++; return m_transmissionId++;
} }
std::mutex m_pendingCallMutex; std::mutex m_pendingCallMutex;
std::vector<jus::FutureBase> m_pendingCall; std::vector<std::pair<uint64_t, jus::FutureBase>> m_pendingCall;
protected: protected:
enum jus::connectionMode m_interfaceMode; enum jus::connectionMode m_interfaceMode;
public: public:
@ -105,7 +106,6 @@ namespace jus {
public: // section call direct public: // section call direct
template<class... _ARGS> template<class... _ARGS>
jus::FutureBase call(const std::string& _functionName, _ARGS&&... _args) { jus::FutureBase call(const std::string& _functionName, _ARGS&&... _args) {
return m_interfaceClient.call(_functionName, _args...);
uint16_t id = getId(); uint16_t id = getId();
std::vector<jus::ActionAsyncClient> asyncAction; std::vector<jus::ActionAsyncClient> asyncAction;
if (m_interfaceMode == jus::connectionMode::modeJson) { if (m_interfaceMode == jus::connectionMode::modeJson) {
@ -118,7 +118,6 @@ namespace jus {
} }
template<class... _ARGS> template<class... _ARGS>
jus::FutureBase callAction(const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) { jus::FutureBase callAction(const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) {
return m_interfaceClient.callAction(_functionName, _args..., _callback);
uint16_t id = getId(); uint16_t id = getId();
std::vector<jus::ActionAsyncClient> asyncAction; std::vector<jus::ActionAsyncClient> asyncAction;
if (m_interfaceMode == jus::connectionMode::modeJson) { if (m_interfaceMode == jus::connectionMode::modeJson) {
@ -132,31 +131,49 @@ namespace jus {
public: // section call with service ID / Client ID public: // section call with service ID / Client ID
template<class... _ARGS> template<class... _ARGS>
jus::FutureBase call(const std::string& _functionName, _ARGS&&... _args) { jus::FutureBase callService(uint32_t _serviceId, const std::string& _functionName, _ARGS&&... _args) {
return m_clientInterface->m_interfaceClient.callService(m_serviceId, _functionName, _args...);
uint16_t id = getId(); uint16_t id = getId();
std::vector<jus::ActionAsyncClient> asyncActionToDo; std::vector<jus::ActionAsyncClient> asyncActionToDo;
if (m_interfaceMode == jus::connectionMode::modeJson) { if (m_interfaceMode == jus::connectionMode::modeJson) {
ejson::Object callElem = jus::createCallService(asyncActionToDo, id, m_serviceId, _functionName, std::forward<_ARGS>(_args)...); ejson::Object callElem = jus::createCallService(asyncActionToDo, id, _serviceId, _functionName, std::forward<_ARGS>(_args)...);
return callJson(id, callElem, asyncActionToDo); return callJson(id, callElem, asyncActionToDo);
} else { } else {
jus::Buffer callElem = jus::createBinaryCallService(asyncActionToDo, id, m_serviceId, _functionName, std::forward<_ARGS>(_args)...); jus::Buffer callElem = jus::createBinaryCallService(asyncActionToDo, id, _serviceId, _functionName, std::forward<_ARGS>(_args)...);
return callBinary(id, callElem, asyncActionToDo); return callBinary(id, callElem, asyncActionToDo);
} }
} }
template<class... _ARGS> template<class... _ARGS>
jus::FutureBase callAction(const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) { jus::FutureBase callServiceAction(uint32_t _serviceId, const std::string& _functionName, _ARGS&&... _args, jus::FutureData::ObserverFinish _callback) {
return m_clientInterface->m_interfaceClient.callServiceAction(m_serviceId, _functionName, _args..., _callback);
uint16_t id = getId(); uint16_t id = getId();
std::vector<jus::ActionAsyncClient> asyncActionToDo; std::vector<jus::ActionAsyncClient> asyncActionToDo;
if (m_interfaceMode == jus::connectionMode::modeJson) { if (m_interfaceMode == jus::connectionMode::modeJson) {
ejson::Object callElem = jus::createCallService(asyncActionToDo, id, m_serviceId, _functionName, std::forward<_ARGS>(_args)...); ejson::Object callElem = jus::createCallService(asyncActionToDo, id, _serviceId, _functionName, std::forward<_ARGS>(_args)...);
return callJson(id, callElem, asyncActionToDo, _callback); return callJson(id, callElem, asyncActionToDo, _callback);
} else { } else {
jus::Buffer callElem = jus::createBinaryCallService(asyncActionToDo, id, m_serviceId, _functionName, std::forward<_ARGS>(_args)...); jus::Buffer callElem = jus::createBinaryCallService(asyncActionToDo, id, _serviceId, _functionName, std::forward<_ARGS>(_args)...);
return callBinary(id, callElem, asyncActionToDo, _callback); return callBinary(id, callElem, asyncActionToDo, _callback);
} }
} }
template<class... _ARGS>
jus::FutureBase callClient(uint32_t _clientId,
const std::string& _functionName,
_ARGS&&... _args) {
return callService(_clientId, _functionName, _args...);
}
template<class... _ARGS>
jus::FutureBase callClientAction(uint32_t _clientId,
const std::string& _functionName,
_ARGS&&... _args,
jus::FutureData::ObserverFinish _callback) {
return callServiceAction(_clientId, _functionName, _args..., _callback);
}
jus::FutureBase callForward(uint32_t _clientId,
jus::Buffer& _Buffer,
uint64_t _singleReferenceId,
jus::FutureData::ObserverFinish _callback);
void callForwardMultiple(uint32_t _clientId,
jus::Buffer& _Buffer,
uint64_t _singleReferenceId);
public: // answers ... public: // answers ...
void answerProtocolError(uint32_t _transactionId, const std::string& _errorHelp); void answerProtocolError(uint32_t _transactionId, const std::string& _errorHelp);