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

[DEV] start communicate with the binary mode

Browse Source
This commit is contained in:
Edouard DUPIN 2016-06-11 00:08:26 +02:00
parent 4983fea721
commit 232954f689
8 changed files with 239 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

145
jus/Buffer.cpp
View File

@ -42,9 +42,9 @@ jus::Buffer::Buffer() {
clear();
}
void jus::Buffer::composeWith(const std::vector<uint8_t>& _buffer) {
void jus::Buffer::internalComposeWith(const uint8_t* _buffer, uint32_t _lenght) {
clear();
m_header.lenght = _buffer.size();
m_header.lenght = _lenght;
uint32_t offset = 0;
memcpy(reinterpret_cast<char*>(&m_header) + sizeof(uint32_t), &_buffer[offset], sizeof(headerBin)-sizeof(uint32_t));
offset += sizeof(headerBin)-sizeof(uint32_t);
@ -52,7 +52,7 @@ void jus::Buffer::composeWith(const std::vector<uint8_t>& _buffer) {
m_paramOffset.resize(m_header.numberOfParameter);
memcpy(&m_paramOffset[0], &_buffer[offset], m_header.numberOfParameter * sizeof(uint16_t));
offset += m_header.numberOfParameter * sizeof(uint16_t);
m_data.resize(_buffer.size() - offset);
m_data.resize(_lenght - offset);
memcpy(&m_data[0], &_buffer[offset], m_data.size());
} else {
// TODO : check size ...
@ -60,6 +60,14 @@ void jus::Buffer::composeWith(const std::vector<uint8_t>& _buffer) {
JUS_INFO("Get binary messages " << generateHumanString());
}
void jus::Buffer::composeWith(const std::vector<uint8_t>& _buffer) {
internalComposeWith(&_buffer[0], _buffer.size());
}
void jus::Buffer::composeWith(const std::string& _buffer) {
internalComposeWith(reinterpret_cast<const uint8_t*>(&_buffer[0]), _buffer.size());
}
void jus::Buffer::clear() {
JUS_WARNING("clear buffer");
m_data.clear();
@ -85,13 +93,14 @@ std::string jus::Buffer::generateHumanString() {
switch (type) {
case jus::Buffer::typeMessage::call:
out += " nbParam=" + etk::to_string(getNumberParameter());
out += " call='" + getCall() + "'";
break;
case jus::Buffer::typeMessage::answer:
if (getNumberParameter() == 1) {
if (m_paramOffset.size() == 1) {
out += " mode=Value";
} else if (getNumberParameter() == 2) {
} else if (m_paramOffset.size() == 2) {
out += " mode=Error";
} else if (getNumberParameter() == 3) {
} else if (m_paramOffset.size() == 3) {
out += " mode=Value+Error";
} else {
out += " mode=???";
@ -228,6 +237,7 @@ uint32_t jus::Buffer::internalGetParameterSize(int32_t _id) const {
startPos += strlen(type) + 1;
// get real data size
out = endPos - startPos;
out --;
if (out < 0) {
JUS_ERROR("Get size < 0 : " << out);
out = 0;
@ -264,6 +274,41 @@ void jus::Buffer::addParameter<std::string>(const std::string& _value) {
memcpy(&m_data[currentOffset], &_value[0], _value.size());
}
template<>
void jus::Buffer::addParameter<std::vector<std::string>>(const std::vector<std::string>& _value) {
int32_t currentOffset = m_data.size();
m_paramOffset.push_back(currentOffset);
m_data.push_back('v');
m_data.push_back('e');
m_data.push_back('c');
m_data.push_back('t');
m_data.push_back('o');
m_data.push_back('r');
m_data.push_back(':');
m_data.push_back('s');
m_data.push_back('t');
m_data.push_back('r');
m_data.push_back('i');
m_data.push_back('n');
m_data.push_back('g');
m_data.push_back('\0');
// count all datas:
uint32_t size = 0;
for (auto &it : _value) {
size+=it.size()+1;
}
uint16_t nb = _value.size();
currentOffset = m_data.size();
m_data.resize(m_data.size()+size+2);
memcpy(&m_data[currentOffset], &nb, sizeof(uint16_t));
currentOffset += sizeof(uint16_t);
for (auto &it : _value) {
memcpy(&m_data[currentOffset], &it[0], it.size());
currentOffset += it.size();
m_data[currentOffset] = '\0';
currentOffset++;
}
}
template<>
void jus::Buffer::addParameter<int8_t>(const int8_t& _value) {
int32_t currentOffset = m_data.size();
m_paramOffset.push_back(currentOffset);
@ -418,49 +463,39 @@ void jus::Buffer::addParameter<bool>(const bool& _value) {
}
}
template<>
bool jus::Buffer::internalGetParameter<bool>(int32_t _id) const {
std::string type = internalGetParameterType(_id);
const uint8_t* pointer = internalGetParameterPointer(_id);
uint32_t dataSize = internalGetParameterSize(_id);
if (createType<bool>() != type) {
return 0;
}
const char* pointer2 = reinterpret_cast<const char*>(pointer);
if (*pointer2 == 'T') {
return true;
}
return false;
}
template<>
std::string jus::Buffer::internalGetParameter<std::string>(int32_t _id) const {
std::string out;
if (m_paramOffset.size() <= _id) {
JUS_ERROR("out of range Id for parameter ... " << _id << " have " << m_paramOffset.size());
return out;
}
int32_t startPos = m_paramOffset[_id];
int32_t endPos = m_data.size();
if (m_paramOffset.size() > _id+1) {
endPos = m_paramOffset[_id+1];
}
// First get type:
const char* type = reinterpret_cast<const char*>(&m_data[startPos]); // Will be stop with \0 ...
if (strcmp(type, "string") == 0) {
// OK
// move in the buffer pos
startPos += strlen(type) + 1;
// get real data size
int32_t dataSize = endPos - startPos;
if (dataSize < 0) {
JUS_ERROR("Get size < 0 : " << dataSize);
} else if (dataSize < 0) {
// nothing to do ...
} else {
// Allocate data
out.resize(dataSize, ' ');
memcpy(&out[0], &m_data[startPos], dataSize);
}
} else {
//wrong type ...
JUS_ERROR("Can not convert '" << type << "' into 'string'");
}
std::string type = internalGetParameterType(_id);
const uint8_t* pointer = internalGetParameterPointer(_id);
uint32_t dataSize = internalGetParameterSize(_id);
out.resize(dataSize, 0);
memcpy(&out[0], pointer, out.size());
return out;
}
template<>
int8_t jus::Buffer::internalGetParameter<int8_t>(int32_t _id) const {
std::string type = getParameterType(_id);
const uint8_t* pointer = getParameterPointer(_id);
uint32_t dataSize = getParameterSize(_id);
std::string type = internalGetParameterType(_id);
const uint8_t* pointer = internalGetParameterPointer(_id);
uint32_t dataSize = internalGetParameterSize(_id);
if (createType<int8_t>() != type) {
return 0;
}
@ -468,11 +503,23 @@ int8_t jus::Buffer::internalGetParameter<int8_t>(int32_t _id) const {
return *pointer2;
}
template<>
int16_t jus::Buffer::internalGetParameter<int16_t>(int32_t _id) const {
std::string type = internalGetParameterType(_id);
const uint8_t* pointer = internalGetParameterPointer(_id);
uint32_t dataSize = internalGetParameterSize(_id);
if (createType<int16_t>() != type) {
return 0;
}
const int16_t* pointer2 = reinterpret_cast<const int16_t*>(pointer);
return *pointer2;
}
template<>
int32_t jus::Buffer::internalGetParameter<int32_t>(int32_t _id) const {
std::string type = getParameterType(_id);
const uint8_t* pointer = getParameterPointer(_id);
uint32_t dataSize = getParameterSize(_id);
std::string type = internalGetParameterType(_id);
const uint8_t* pointer = internalGetParameterPointer(_id);
uint32_t dataSize = internalGetParameterSize(_id);
if (createType<int32_t>() != type) {
return 0;
}
@ -480,6 +527,18 @@ int32_t jus::Buffer::internalGetParameter<int32_t>(int32_t _id) const {
return *pointer2;
}
template<>
int64_t jus::Buffer::internalGetParameter<int64_t>(int32_t _id) const {
std::string type = internalGetParameterType(_id);
const uint8_t* pointer = internalGetParameterPointer(_id);
uint32_t dataSize = internalGetParameterSize(_id);
if (createType<int64_t>() != type) {
return 0;
}
const int64_t* pointer2 = reinterpret_cast<const int64_t*>(pointer);
return *pointer2;
}
void jus::Buffer::addError(const std::string& _value, const std::string& _comment) {
addParameter(_value);

8
jus/Buffer.h
View File

@ -105,9 +105,12 @@ namespace jus {
uint32_t getDataSize() const {
return m_data.size();
}
private:
void internalComposeWith(const uint8_t* _buffer, uint32_t _lenght);
public:
Buffer();
void composeWith(const std::vector<uint8_t>& _buffer);
void composeWith(const std::string& _buffer);
std::string generateHumanString();
void clear();
uint16_t getProtocalVersion() const;
@ -170,6 +173,11 @@ namespace jus {
void addAnswer(const JUS_TYPE_DATA& _value) {
addParameter(_value);
}
// TODO : Do it better check error ... ==> can be good ...
template<class JUS_TYPE_DATA>
JUS_TYPE_DATA getAnswer() const {
return internalGetParameter<JUS_TYPE_DATA>(0);
}
void addError(const std::string& _value, const std::string& _comment);

77
jus/Client.cpp
View File

@ -22,6 +22,69 @@ jus::Client::~Client() {
}
void jus::Client::onClientDataRaw(jus::Buffer& _value) {
JUS_DEBUG("Get answer RAW : "/* << _value*/);
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;
}
}
}
void jus::Client::onClientData(std::string _value) {
JUS_DEBUG("Get answer : " << _value);
ejson::Object obj(_value);
@ -147,20 +210,30 @@ bool jus::Client::connect(const std::string& _remoteUserToConnect){
m_interfaceClient.setInterface(std::move(connection));
m_interfaceClient.connect();
// Force mode binary:
JUS_WARNING("Request change in mode Binary");
jus::Future<bool> retBin = call("setMode", "BIN").wait();
if (retBin.get() == true) {
JUS_WARNING(" ==> accepted binary");
m_interfaceMode = jus::connectionMode::modeBinary;
m_interfaceClient.connectClean();
m_interfaceClient.connectRaw(this, &jus::Client::onClientDataRaw);
JUS_INFO("Connection jump in BINARY ...");
} else {
// stay in JSON
}
JUS_WARNING("Request connect user " << _remoteUserToConnect);
jus::Future<bool> ret = call("connectToUser", _remoteUserToConnect, "jus-client");
ret.wait();
if (ret.hasError() == true) {
JUS_WARNING("Can not connect to user named: '" << _remoteUserToConnect << "' ==> return error");
return false;
}
if (ret.get() == true) {
JUS_WARNING(" ==> accepted connection");
} else {
JUS_WARNING(" ==> Refuse connection");
}
return ret.get();
}
@ -317,3 +390,7 @@ jus::FutureBase jus::Client::callBinary(uint64_t _transactionId,
JUS_VERBOSE("Send Binary [STOP]");
return tmpFuture;
}

1
jus/Client.h
View File

@ -47,6 +47,7 @@ namespace jus {
//client1.authentificate("coucou");
private:
void onClientData(std::string _value);
void onClientDataRaw(jus::Buffer& _value);
std::string asyncRead();
jus::FutureBase callJson(uint64_t _transactionId,
ejson::Object _obj,

10
jus/Future.cpp
View File

@ -15,6 +15,8 @@ namespace jus {
if (m_data == nullptr) {
return false;
}
// JSON mode
if (m_data->m_mode == false) {
ejson::Value val = m_data->m_returnData["return"];
if (val.exist() == false) {
JUS_WARNING("No Return value ...");
@ -26,11 +28,16 @@ namespace jus {
}
return val.toBoolean().get();
}
// binary mode:
return m_data->m_returnDataBinary.getAnswer<bool>();
}
template<>
int64_t jus::Future<int64_t>::get() {
if (m_data == nullptr) {
return 0;
}
// JSON mode
if (m_data->m_mode == false) {
ejson::Value val = m_data->m_returnData["return"];
if (val.exist() == false) {
JUS_WARNING("No Return value ...");
@ -42,6 +49,9 @@ namespace jus {
}
return int64_t(val.toNumber().get());
}
// binary mode:
return m_data->m_returnDataBinary.getAnswer<bool>();
}
template<>
int32_t jus::Future<int32_t>::get() {
if (m_data == nullptr) {

5
jus/FutureBase.cpp
View File

@ -163,13 +163,12 @@ bool jus::FutureBase::setAnswer(const jus::Buffer& _returnValue) {
// finish is false ==> normal case ...
}
return false;
}
m_data->m_returnData = _returnValue;
}*/
m_data->m_returnDataBinary = _returnValue;
m_data->m_isFinished = true;
if (m_data->m_callbackFinish != nullptr) {
return m_data->m_callbackFinish(*this);
}
*/
return true;
}
void jus::FutureBase::setSynchronous() {

20
jus/TcpString.cpp
View File

@ -39,7 +39,7 @@ jus::TcpString::~TcpString() {
void jus::TcpString::setInterfaceName(const std::string& _name) {
ethread::setName(*m_thread, "Tcp-" + _name);
}
// TODO : Do it better :
void jus::TcpString::threadCallback() {
ethread::setName("TcpString-input");
// get datas:
@ -47,15 +47,29 @@ void jus::TcpString::threadCallback() {
&& m_connection.getConnectionStatus() == enet::Tcp::status::link) {
// READ section data:
if (m_observerElement != nullptr) {
JUS_PRINT("Call String ...");
std::string data = std::move(read());
JUS_VERBOSE("Receive data: '" << data << "'");
JUS_PRINT("Receive data: '" << data << "'");
if (data.size() != 0) {
m_lastReceive = std::chrono::steady_clock::now();
JUS_PRINT(" Call function ... 1");
if (m_observerElement != nullptr) {
m_observerElement(std::move(data));
} else if (m_observerRawElement != nullptr) {
jus::Buffer dataRaw;
dataRaw.composeWith(data);
m_observerRawElement(dataRaw);
} else {
JUS_ERROR("Lose DATA ...");
}
JUS_PRINT(" Call function ... 1 (done)");
}
} else if (m_observerRawElement != nullptr) {
jus::Buffer data = std::move(readRaw());
JUS_PRINT("Call Raw ...");
jus::Buffer data = readRaw();
JUS_PRINT(" Call function ... 2");
m_observerRawElement(data);
JUS_PRINT(" Call function ... 2 (done)");
}
}
m_threadRunning = false;

1
jus/debug.h
View File

@ -13,6 +13,7 @@ namespace jus {
#define JUS_BASE(info,data) ELOG_BASE(jus::getLogId(),info,data)
#define JUS_PRINT(data) JUS_BASE(-1, data)
#define JUS_CRITICAL(data) JUS_BASE(1, data)
#define JUS_ERROR(data) JUS_BASE(2, data)
#define JUS_WARNING(data) JUS_BASE(3, data)