discovery_mq->Disconnect();
}
-void AittDiscovery::UpdateDiscoveryMsg(const std::string &protocol, const void *msg, size_t length)
+void AittDiscovery::UpdateDiscoveryMsg(const std::string &protocol, const void *msg, int length)
{
auto it = discovery_map.find(protocol);
if (it == discovery_map.end())
true);
}
-AittDiscovery::DiscoveryBlob::DiscoveryBlob(const void *msg, size_t length)
+AittDiscovery::DiscoveryBlob::DiscoveryBlob(const void *msg, int length)
: len(length), data(new char[len])
{
memcpy(data.get(), msg, length);
const std::string &password);
void Restart();
void Stop();
- void UpdateDiscoveryMsg(const std::string &protocol, const void *msg, size_t length);
+ void UpdateDiscoveryMsg(const std::string &protocol, const void *msg, int length);
int AddDiscoveryCB(const std::string &protocol, const DiscoveryCallback &cb);
void RemoveDiscoveryCB(int callback_id);
bool CompareTopic(const std::string &left, const std::string &right);
private:
struct DiscoveryBlob {
- explicit DiscoveryBlob(const void *msg, size_t length);
+ explicit DiscoveryBlob(const void *msg, int length);
~DiscoveryBlob();
DiscoveryBlob(const DiscoveryBlob &src);
DiscoveryBlob &operator=(const DiscoveryBlob &src);
- size_t len;
+ int len;
std::shared_ptr<char> data;
};
typedef void *(
*ModuleEntry)(AittProtocol type, AittDiscovery &discovery, const std::string &my_ip);
using SubscribeCallback = std::function<void(const std::string &topic, const void *msg,
- const size_t szmsg, void *cbdata, const std::string &correlation)>;
+ const int szmsg, void *cbdata, const std::string &correlation)>;
static constexpr const char *const MODULE_ENTRY_NAME = DEFINE_TO_STR(AITT_TRANSPORT_NEW);
}
virtual ~AittTransport(void) = default;
- virtual void Publish(const std::string &topic, const void *data, const size_t datalen,
+ virtual void Publish(const std::string &topic, const void *data, const int datalen,
const std::string &correlation, AittQoS qos = AITT_QOS_AT_MOST_ONCE,
bool retain = false) = 0;
-
- virtual void Publish(const std::string &topic, const void *data, const size_t datalen,
+ virtual void Publish(const std::string &topic, const void *data, const int datalen,
AittQoS qos = AITT_QOS_AT_MOST_ONCE, bool retain = false) = 0;
-
virtual void *Subscribe(const std::string &topic, const SubscribeCallback &cb,
void *cbdata = nullptr, AittQoS qos = AITT_QOS_AT_MOST_ONCE) = 0;
- virtual void *Subscribe(const std::string &topic, const SubscribeCallback &cb, const void *data,
- const size_t datalen, void *cbdata = nullptr, AittQoS qos = AITT_QOS_AT_MOST_ONCE) = 0;
-
virtual void *Unsubscribe(void *handle) = 0;
+
AittProtocol GetProtocol() { return protocol; }
protected:
class API AITT {
public:
using SubscribeCallback =
- std::function<void(MSG *msg, const void *data, const size_t datalen, void *user_data)>;
+ std::function<void(MSG *msg, const void *data, const int datalen, void *user_data)>;
using ConnectionCallback = std::function<void(AITT &, int, void *user_data)>;
explicit AITT(const std::string &id, const std::string &ip_addr,
AittOption option = AittOption(false, false));
virtual ~AITT(void);
- void SetWillInfo(const std::string &topic, const void *data, const size_t datalen, AittQoS qos,
+ void SetWillInfo(const std::string &topic, const void *data, const int datalen, AittQoS qos,
bool retain);
void SetConnectionCallback(ConnectionCallback cb, void *user_data = nullptr);
void Connect(const std::string &host = AITT_LOCALHOST, int port = AITT_PORT,
const std::string &username = std::string(), const std::string &password = std::string());
void Disconnect(void);
- void Publish(const std::string &topic, const void *data, const size_t datalen,
+ void Publish(const std::string &topic, const void *data, const int datalen,
AittProtocol protocols = AITT_TYPE_MQTT, AittQoS qos = AITT_QOS_AT_MOST_ONCE,
bool retain = false);
- int PublishWithReply(const std::string &topic, const void *data, const size_t datalen,
+ int PublishWithReply(const std::string &topic, const void *data, const int datalen,
AittProtocol protocol, AittQoS qos, bool retain, const SubscribeCallback &cb,
void *cbdata, const std::string &correlation);
- int PublishWithReplySync(const std::string &topic, const void *data, const size_t datalen,
+ int PublishWithReplySync(const std::string &topic, const void *data, const int datalen,
AittProtocol protocol, AittQoS qos, bool retain, const SubscribeCallback &cb,
void *cbdata, const std::string &correlation, int timeout_ms = 0);
AittQoS qos = AITT_QOS_AT_MOST_ONCE);
void *Unsubscribe(AittSubscribeID handle);
- void SendReply(MSG *msg, const void *data, const size_t datalen, bool end = true);
+ void SendReply(MSG *msg, const void *data, const int datalen, bool end = true);
AittStream *CreateStream(AittStreamProtocol type, const std::string &topic,
AittStreamRole role);
typedef void *(*ModuleEntry)(const char *id, const AittOption &option);
using SubscribeCallback = std::function<void(MSG *msg, const std::string &topic,
- const void *data, const size_t datalen, void *user_data)>;
+ const void *data, const int datalen, void *user_data)>;
using MQConnectionCallback = std::function<void(int)>;
static constexpr const char *const MODULE_ENTRY_NAME = DEFINE_TO_STR(AITT_MQ_NEW);
virtual void SetConnectionCallback(const MQConnectionCallback &cb) = 0;
virtual void Connect(const std::string &host, int port, const std::string &username,
const std::string &password) = 0;
- virtual void SetWillInfo(const std::string &topic, const void *msg, size_t szmsg, int qos,
+ virtual void SetWillInfo(const std::string &topic, const void *msg, int szmsg, int qos,
bool retain) = 0;
virtual void Disconnect(void) = 0;
- virtual void Publish(const std::string &topic, const void *data, const size_t datalen,
- int qos = 0, bool retain = false) = 0;
- virtual void PublishWithReply(const std::string &topic, const void *data, const size_t datalen,
+ virtual void Publish(const std::string &topic, const void *data, const int datalen, int qos = 0,
+ bool retain = false) = 0;
+ virtual void PublishWithReply(const std::string &topic, const void *data, const int datalen,
int qos, bool retain, const std::string &reply_topic, const std::string &correlation) = 0;
- virtual void SendReply(MSG *msg, const void *data, const size_t datalen, int qos,
- bool retain) = 0;
+ virtual void SendReply(MSG *msg, const void *data, const int datalen, int qos, bool retain) = 0;
virtual void *Subscribe(const std::string &topic, const SubscribeCallback &cb,
void *user_data = nullptr, int qos = 0) = 0;
virtual void *Unsubscribe(void *handle) = 0;
* @see aitt_subscribe()
* @see aitt_subscribe_full()
*/
-typedef void (
- *aitt_sub_fn)(aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data);
+typedef void (*aitt_sub_fn)(aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data);
/**
* @brief Create a new AITT service instance.
*
* @see aitt_connect()
*/
-int aitt_will_set(aitt_h handle, const char *topic, const void *msg, const size_t msg_len,
+int aitt_will_set(aitt_h handle, const char *topic, const void *msg, const int msg_len,
aitt_qos_e qos, bool retain);
/**
* @retval #AITT_ERROR_INVALID_PARAMETER Invalid parameter
* @retval #AITT_ERROR_SYSTEM System errors
*/
-int aitt_publish(aitt_h handle, const char *topic, const void *msg, const size_t msg_len);
+int aitt_publish(aitt_h handle, const char *topic, const void *msg, const int msg_len);
/**
* @brief Publish a message on a given topic as aitt_publish(), but takes protocols and qos.
* @retval #AITT_ERROR_INVALID_PARAMETER Invalid parameter
* @retval #AITT_ERROR_SYSTEM System errors
*/
-int aitt_publish_full(aitt_h handle, const char *topic, const void *msg, const size_t msg_len,
+int aitt_publish_full(aitt_h handle, const char *topic, const void *msg, const int msg_len,
int protocols, aitt_qos_e qos);
/**
* @retval #AITT_ERROR_INVALID_PARAMETER Invalid parameter
* @retval #AITT_ERROR_SYSTEM System errors
*/
-int aitt_publish_with_reply(aitt_h handle, const char *topic, const void *msg, const size_t msg_len,
+int aitt_publish_with_reply(aitt_h handle, const char *topic, const void *msg, const int msg_len,
aitt_protocol_e protocols, aitt_qos_e qos, const char *correlation, aitt_sub_fn cb,
void *user_data);
* @retval #AITT_ERROR_INVALID_PARAMETER Invalid parameter
* @retval #AITT_ERROR_SYSTEM System errors
*/
-int aitt_send_reply(aitt_h handle, aitt_msg_h msg_handle, const void *reply, const size_t reply_len,
+int aitt_send_reply(aitt_h handle, aitt_msg_h msg_handle, const void *reply, const int reply_len,
bool end);
/**
DBG_HEX_DUMP(iv_.data(), iv_.size());
}
-size_t AESEncryptor::GetCryptogramSize(size_t plain_size)
+int AESEncryptor::GetCryptogramSize(int plain_size)
{
const int BLOCKSIZE = 16;
return (plain_size / BLOCKSIZE + 1) * BLOCKSIZE;
}
}
-size_t AESEncryptor::Encrypt(const unsigned char *plaintext, int plaintext_len,
+int AESEncryptor::Encrypt(const unsigned char *plaintext, int plaintext_len,
unsigned char *ciphertext)
{
int len;
return ciphertext_len + len;
}
-size_t AESEncryptor::Decrypt(const unsigned char *ciphertext, int ciphertext_len,
+int AESEncryptor::Decrypt(const unsigned char *ciphertext, int ciphertext_len,
unsigned char *plaintext)
{
int len;
static void GenerateKey(unsigned char (&key)[AITT_TCP_ENCRYPTOR_KEY_LEN],
unsigned char (&iv)[AITT_TCP_ENCRYPTOR_IV_LEN]);
void Init(const unsigned char *key, const unsigned char *iv);
- size_t GetCryptogramSize(size_t plain_size);
- size_t Encrypt(const unsigned char *plaintext, int plaintext_len, unsigned char *ciphertext);
- size_t Decrypt(const unsigned char *ciphertext, int ciphertext_len, unsigned char *plaintext);
+ int GetCryptogramSize(int plain_size);
+ int Encrypt(const unsigned char *plaintext, int plaintext_len, unsigned char *ciphertext);
+ int Decrypt(const unsigned char *ciphertext, int ciphertext_len, unsigned char *plaintext);
private:
std::vector<unsigned char> key_;
main_loop.Run();
}
-void Module::Publish(const std::string &topic, const void *data, const size_t datalen,
+void Module::Publish(const std::string &topic, const void *data, const int datalen,
const std::string &correlation, AittQoS qos, bool retain)
{
// NOTE:
}
try {
- size_t length = topic.length();
+ int32_t length = topic.length();
portIt->second->SendSizedData(topic.c_str(), length);
length = datalen;
portIt->second->SendSizedData(data, length);
} // publishTable
}
-void Module::Publish(const std::string &topic, const void *data, const size_t datalen, AittQoS qos,
+void Module::Publish(const std::string &topic, const void *data, const int datalen, AittQoS qos,
bool retain)
{
Publish(topic, data, datalen, std::string(), qos, retain);
return reinterpret_cast<void *>(handle);
}
-void *Module::Subscribe(const std::string &topic, const AittTransport::SubscribeCallback &cb,
- const void *data, const size_t datalen, void *cbdata, AittQoS qos)
-{
- return nullptr;
-}
-
void *Module::Unsubscribe(void *handlePtr)
{
int handle = static_cast<int>(reinterpret_cast<intptr_t>(handlePtr));
return impl->HandleClientDisconnect(handle);
}
- size_t szmsg = 0;
+ int32_t szmsg = 0;
char *msg = nullptr;
std::string topic;
std::string Module::GetTopicName(Module::TCPData *tcp_data)
{
- size_t topic_length = 0;
+ int32_t topic_length = 0;
void *topic_data = nullptr;
int ret = tcp_data->client->RecvSizedData(&topic_data, topic_length);
if (ret < 0) {
}
std::string topic = std::string(static_cast<char *>(topic_data), topic_length);
- INFO("Complete topic = [%s], topic_len = %zu", topic.c_str(), topic_length);
+ INFO("Complete topic = [%s], topic_len = %d", topic.c_str(), topic_length);
free(topic_data);
return topic;
explicit Module(AittProtocol type, AittDiscovery &discovery, const std::string &ip);
virtual ~Module(void);
- void Publish(const std::string &topic, const void *data, const size_t datalen,
+ void Publish(const std::string &topic, const void *data, const int datalen,
const std::string &correlation, AittQoS qos = AITT_QOS_AT_MOST_ONCE,
bool retain = false) override;
-
- void Publish(const std::string &topic, const void *data, const size_t datalen,
+ void Publish(const std::string &topic, const void *data, const int datalen,
AittQoS qos = AITT_QOS_AT_MOST_ONCE, bool retain = false) override;
void *Subscribe(const std::string &topic, const SubscribeCallback &cb, void *cbdata = nullptr,
AittQoS qos = AITT_QOS_AT_MOST_ONCE) override;
-
- void *Subscribe(const std::string &topic, const SubscribeCallback &cb, const void *data,
- const size_t datalen, void *cbdata = nullptr,
- AittQoS qos = AITT_QOS_AT_MOST_ONCE) override;
void *Unsubscribe(void *handle) override;
private:
}
}
-void TCP::Send(const void *data, size_t &szData)
+void TCP::Send(const void *data, int32_t &data_size)
{
- size_t sent = 0;
- while (sent < szData) {
- int ret = send(handle, static_cast<const char *>(data) + sent, szData - sent, 0);
+ int32_t sent = 0;
+ while (sent < data_size) {
+ int ret = send(handle, static_cast<const char *>(data) + sent, data_size - sent, 0);
if (ret < 0) {
- ERR("Fail to send data, handle = %d, size = %zu", handle, szData);
+ ERR("Fail to send data, handle = %d, size = %d", handle, data_size);
throw std::runtime_error(strerror(errno));
}
sent += ret;
}
- szData = sent;
+ data_size = sent;
}
-void TCP::SendSizedData(const void *data, size_t &szData)
+void TCP::SendSizedData(const void *data, int32_t &szData)
{
if (secure)
SendSizedDataSecure(data, szData);
SendSizedDataNormal(data, szData);
}
-int TCP::Recv(void *data, size_t &szData)
+int TCP::Recv(void *data, int32_t &data_size)
{
- size_t received = 0;
- while (received < szData) {
- int ret = recv(handle, static_cast<char *>(data) + received, szData - received, 0);
+ int32_t received = 0;
+ while (received < data_size) {
+ int ret = recv(handle, static_cast<char *>(data) + received, data_size - received, 0);
if (ret < 0) {
- ERR("Fail to recv data, handle = %d, size = %zu", handle, szData);
+ ERR("Fail to recv data, handle = %d, size = %d", handle, data_size);
throw std::runtime_error(strerror(errno));
}
if (ret == 0) {
received += ret;
}
- szData = received;
+ data_size = received;
return 0;
}
-int TCP::RecvSizedData(void **data, size_t &szData)
+int TCP::RecvSizedData(void **data, int32_t &szData)
{
if (secure)
return RecvSizedDataSecure(data, szData);
return RecvSizedDataNormal(data, szData);
}
-int TCP::HandleZeroMsg(void **data, size_t &data_size)
+int TCP::HandleZeroMsg(void **data, int32_t &data_size)
{
// distinguish between connection problems and zero-size messages
INFO("Got a zero-size message.");
return ntohs(addr.sin_port);
}
-void TCP::SendSizedDataNormal(const void *data, size_t &data_size)
+void TCP::SendSizedDataNormal(const void *data, int32_t &data_size)
{
- size_t fixed_data_size = data_size;
+ int32_t fixed_data_size = data_size;
if (0 == data_size) {
// distinguish between connection problems and zero-size messages
INFO("Send a zero-size message.");
- fixed_data_size = UINT32_MAX;
+ fixed_data_size = INT32_MAX;
}
- size_t size_len = sizeof(fixed_data_size);
+ int32_t size_len = sizeof(fixed_data_size);
Send(static_cast<const void *>(&fixed_data_size), size_len);
Send(data, data_size);
}
-int TCP::RecvSizedDataNormal(void **data, size_t &data_size)
+int TCP::RecvSizedDataNormal(void **data, int32_t &data_size)
{
int ret;
- size_t data_len = 0;
- size_t size_len = sizeof(data_len);
+ int32_t data_len = 0;
+ int32_t size_len = sizeof(data_len);
ret = Recv(static_cast<void *>(&data_len), size_len);
if (ret < 0) {
ERR("Recv() Fail(%d)", ret);
return ret;
}
- if (data_len == UINT32_MAX)
+ if (data_len == INT32_MAX)
return HandleZeroMsg(data, data_size);
if (AITT_MESSAGE_MAX < data_len) {
- ERR("Invalid Size(%zu)", data_len);
+ ERR("Invalid Size(%d)", data_len);
return -1;
}
void *data_buf = malloc(data_len);
return 0;
}
-void TCP::SendSizedDataSecure(const void *data, size_t &data_size)
+void TCP::SendSizedDataSecure(const void *data, int32_t &data_size)
{
- size_t fixed_data_size = data_size;
+ int32_t fixed_data_size = data_size;
if (0 == data_size) {
// distinguish between connection problems and zero-size messages
INFO("Send a zero-size message.");
- fixed_data_size = UINT32_MAX;
+ fixed_data_size = INT32_MAX;
}
- size_t size_len;
+ int32_t size_len;
if (data_size) {
unsigned char data_buf[crypto.GetCryptogramSize(data_size)];
- size_t data_len =
+ int32_t data_len =
crypto.Encrypt(static_cast<const unsigned char *>(data), data_size, data_buf);
- unsigned char size_buf[crypto.GetCryptogramSize(sizeof(size_t))];
+ unsigned char size_buf[crypto.GetCryptogramSize(sizeof(int32_t))];
size_len = crypto.Encrypt((unsigned char *)&data_len, sizeof(data_len), size_buf);
Send(size_buf, size_len);
Send(data_buf, data_len);
} else {
- unsigned char size_buf[crypto.GetCryptogramSize(sizeof(size_t))];
+ unsigned char size_buf[crypto.GetCryptogramSize(sizeof(int32_t))];
size_len =
crypto.Encrypt((unsigned char *)&fixed_data_size, sizeof(fixed_data_size), size_buf);
Send(size_buf, size_len);
}
}
-int TCP::RecvSizedDataSecure(void **data, size_t &data_size)
+int TCP::RecvSizedDataSecure(void **data, int32_t &data_size)
{
int ret;
- size_t cipher_size_len = crypto.GetCryptogramSize(sizeof(size_t));
+ int32_t cipher_size_len = crypto.GetCryptogramSize(sizeof(int32_t));
unsigned char cipher_size_buf[cipher_size_len];
ret = Recv(cipher_size_buf, cipher_size_len);
if (ret < 0) {
}
unsigned char plain_size_buf[cipher_size_len];
- size_t cipher_data_len = 0;
+ int32_t cipher_data_len = 0;
crypto.Decrypt(cipher_size_buf, cipher_size_len, plain_size_buf);
memcpy(&cipher_data_len, plain_size_buf, sizeof(cipher_data_len));
- if (cipher_data_len == UINT32_MAX)
+ if (cipher_data_len == INT32_MAX)
return HandleZeroMsg(data, data_size);
if (AITT_MESSAGE_MAX < cipher_data_len) {
- ERR("Invalid Size(%zu)", cipher_data_len);
+ ERR("Invalid Size(%d)", cipher_data_len);
return -1;
}
unsigned char cipher_data_buf[cipher_data_len];
*/
#pragma once
+#include <stdint.h>
#include <sys/socket.h>
-#include <sys/types.h> /* See NOTES */
#include <string>
TCP(const std::string &host, const ConnectInfo &ConnectInfo);
virtual ~TCP(void);
- void Send(const void *data, size_t &szData);
- void SendSizedData(const void *data, size_t &szData);
- int Recv(void *data, size_t &szData);
- int RecvSizedData(void **data, size_t &szData);
+ void Send(const void *data, int32_t &szData);
+ void SendSizedData(const void *data, int32_t &szData);
+ int Recv(void *data, int32_t &szData);
+ int RecvSizedData(void **data, int32_t &szData);
int GetHandle(void);
unsigned short GetPort(void);
void GetPeerInfo(std::string &host, unsigned short &port);
private:
TCP(int handle, sockaddr *addr, socklen_t addrlen, const ConnectInfo &connect_info);
void SetupOptions(const ConnectInfo &connect_info);
- int HandleZeroMsg(void **data, size_t &data_size);
- void SendSizedDataNormal(const void *data, size_t &data_size);
- int RecvSizedDataNormal(void **data, size_t &data_size);
- void SendSizedDataSecure(const void *data, size_t &data_size);
- int RecvSizedDataSecure(void **data, size_t &data_size);
+ int HandleZeroMsg(void **data, int32_t &data_size);
+ void SendSizedDataNormal(const void *data, int32_t &data_size);
+ int RecvSizedDataNormal(void **data, int32_t &data_size);
+ void SendSizedDataSecure(const void *data, int32_t &data_size);
+ int RecvSizedDataSecure(void **data, int32_t &data_size);
int handle;
socklen_t addrlen;
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include <TCP.h>
-#include <TCPServer.h>
+#include "../TCP.h"
+
#include <getopt.h>
#include <glib.h>
#include <memory>
#include <string>
+#include "../TCPServer.h"
+
//#define _LOG_WITH_TIMESTAMP
#include "aitt_internal.h"
#ifdef _LOG_WITH_TIMESTAMP
char buffer[10];
void *ptr = static_cast<void *>(buffer);
- size_t szData = sizeof(HELLO_STRING);
+ int32_t szData = sizeof(HELLO_STRING);
peer->Recv(ptr, szData);
INFO("Gots[%s]", buffer);
INFO("Assigned client port: %u", client->GetPort());
INFO("Send[%s]", HELLO_STRING);
- size_t szBuffer = sizeof(HELLO_STRING);
+ int32_t szBuffer = sizeof(HELLO_STRING);
client->Send(HELLO_STRING, szBuffer);
char buffer[10];
unsigned char ciphertext[encryptor.GetCryptogramSize(TEST_MESSAGE.size())];
unsigned char plaintext[encryptor.GetCryptogramSize(TEST_MESSAGE.size())];
- size_t len =
- encryptor.Encrypt(reinterpret_cast<const unsigned char *>(TEST_MESSAGE.c_str()),
- TEST_MESSAGE.size(), ciphertext);
+ int len = encryptor.Encrypt(reinterpret_cast<const unsigned char *>(TEST_MESSAGE.c_str()),
+ TEST_MESSAGE.size(), ciphertext);
len = encryptor.Decrypt(ciphertext, len, plaintext);
plaintext[len] = 0;
ASSERT_STREQ(TEST_MESSAGE.c_str(), reinterpret_cast<char *>(plaintext));
char byeBuffer[TEST_BUFFER_SIZE];
customTest = [this, &helloBuffer](void) mutable -> void {
- size_t szData = sizeof(TEST_BUFFER_HELLO);
+ int32_t szData = sizeof(TEST_BUFFER_HELLO);
client->Recv(static_cast<void *>(helloBuffer), szData);
szData = sizeof(TEST_BUFFER_BYE);
RunServer();
- size_t szMsg = sizeof(TEST_BUFFER_HELLO);
+ int32_t szMsg = sizeof(TEST_BUFFER_HELLO);
peer->Send(TEST_BUFFER_HELLO, szMsg);
szMsg = sizeof(TEST_BUFFER_BYE);
}
void Module::Start(void)
-{
+{
}
void Module::SetStateCallback(StateCallback cb, void *user_data)
-DLOG_STDOUT:BOOL=%{stdoutlog} \
-DPLATFORM="tizen" \
-DVERSIONING:BOOL=OFF \
- -DWITH_WEBRTC:BOOL=ON \
+ -DWITH_WEBRTC:BOOL=OFF \
-DWITH_RTSP:BOOL=ON \
-DCMAKE_INSTALL_PREFIX:PATH=%{_prefix} \
-DCMAKE_VERBOSE_MAKEFILE=OFF \
{
}
-void AITT::SetWillInfo(const std::string &topic, const void *data, const size_t datalen,
- AittQoS qos, bool retain)
+void AITT::SetWillInfo(const std::string &topic, const void *data, const int datalen, AittQoS qos,
+ bool retain)
{
return pImpl->SetWillInfo(topic, data, datalen, qos, retain);
}
return pImpl->Disconnect();
}
-void AITT::Publish(const std::string &topic, const void *data, const size_t datalen,
+void AITT::Publish(const std::string &topic, const void *data, const int datalen,
AittProtocol protocols, AittQoS qos, bool retain)
{
if (AITT_PAYLOAD_MAX < datalen) {
- ERR("Invalid Size(%zu)", datalen);
+ ERR("Invalid Size(%d)", datalen);
throw std::runtime_error("Invalid Size");
}
return pImpl->Publish(topic, data, datalen, protocols, qos, retain);
}
-int AITT::PublishWithReply(const std::string &topic, const void *data, const size_t datalen,
+int AITT::PublishWithReply(const std::string &topic, const void *data, const int datalen,
AittProtocol protocol, AittQoS qos, bool retain, const SubscribeCallback &cb, void *cbdata,
const std::string &correlation)
{
if (AITT_PAYLOAD_MAX < datalen) {
- ERR("Invalid Size(%zu)", datalen);
+ ERR("Invalid Size(%d)", datalen);
throw std::runtime_error("Invalid Size");
}
correlation);
}
-int AITT::PublishWithReplySync(const std::string &topic, const void *data, const size_t datalen,
+int AITT::PublishWithReplySync(const std::string &topic, const void *data, const int datalen,
AittProtocol protocol, AittQoS qos, bool retain, const SubscribeCallback &cb, void *cbdata,
const std::string &correlation, int timeout_ms)
{
if (AITT_PAYLOAD_MAX < datalen) {
- ERR("Invalid Size(%zu)", datalen);
+ ERR("Invalid Size(%d)", datalen);
throw std::runtime_error("Invalid Size");
}
return pImpl->Unsubscribe(handle);
}
-void AITT::SendReply(MSG *msg, const void *data, size_t datalen, bool end)
+void AITT::SendReply(MSG *msg, const void *data, int datalen, bool end)
{
if (AITT_PAYLOAD_MAX < datalen) {
- ERR("Invalid Size(%zu)", datalen);
+ ERR("Invalid Size(%d)", datalen);
throw std::runtime_error("Invalid Size");
}
main_loop.Run();
}
-void AITT::Impl::SetWillInfo(const std::string &topic, const void *data, const size_t datalen,
+void AITT::Impl::SetWillInfo(const std::string &topic, const void *data, const int datalen,
AittQoS qos, bool retain)
{
mq->SetWillInfo(topic, data, datalen, qos, retain);
{
}
-void AITT::Impl::Publish(const std::string &topic, const void *data, const size_t datalen,
+void AITT::Impl::Publish(const std::string &topic, const void *data, const int datalen,
AittProtocol protocols, AittQoS qos, bool retain)
{
if ((protocols & AITT_TYPE_MQTT) == AITT_TYPE_MQTT)
return mq->Subscribe(
topic,
[this, handle, loop_handle, cb](MSG *msg, const std::string &topic, const void *data,
- const size_t datalen, void *mq_user_data) {
+ const int datalen, void *mq_user_data) {
void *delivery = malloc(datalen);
if (delivery)
memcpy(delivery, data, datalen);
user_data, qos);
}
-void AITT::Impl::DetachedCB(SubscribeCallback cb, MSG msg, void *data, const size_t datalen,
+void AITT::Impl::DetachedCB(SubscribeCallback cb, MSG msg, void *data, const int datalen,
void *user_data, MainLoopHandler::MainLoopResult result, int fd,
MainLoopHandler::MainLoopData *loop_data)
{
return user_data;
}
-int AITT::Impl::PublishWithReply(const std::string &topic, const void *data, const size_t datalen,
+int AITT::Impl::PublishWithReply(const std::string &topic, const void *data, const int datalen,
AittProtocol protocol, AittQoS qos, bool retain, const SubscribeCallback &cb, void *user_data,
const std::string &correlation)
{
Subscribe(
replyTopic,
- [this, cb](MSG *sub_msg, const void *sub_data, const size_t sub_datalen,
- void *sub_cbdata) {
+ [this, cb](MSG *sub_msg, const void *sub_data, const int sub_datalen, void *sub_cbdata) {
if (sub_msg->IsEndSequence()) {
try {
Unsubscribe(sub_msg->GetID());
return 0;
}
-int AITT::Impl::PublishWithReplySync(const std::string &topic, const void *data,
- const size_t datalen, AittProtocol protocol, AittQoS qos, bool retain,
- const SubscribeCallback &cb, void *user_data, const std::string &correlation, int timeout_ms)
+int AITT::Impl::PublishWithReplySync(const std::string &topic, const void *data, const int datalen,
+ AittProtocol protocol, AittQoS qos, bool retain, const SubscribeCallback &cb, void *user_data,
+ const std::string &correlation, int timeout_ms)
{
std::string replyTopic = topic + RESPONSE_POSTFIX + std::to_string(reply_id++);
subscribe_handle = SubscribeMQ(
info, &sync_loop, replyTopic,
- [&](MSG *sub_msg, const void *sub_data, const size_t sub_datalen, void *sub_cbdata) {
+ [&](MSG *sub_msg, const void *sub_data, const int sub_datalen, void *sub_cbdata) {
if (sub_msg->IsEndSequence()) {
try {
Unsubscribe(sub_msg->GetID());
return modules.Get(handle->first)
.Subscribe(
topic,
- [handle, cb](const std::string &topic, const void *data, const size_t datalen,
+ [handle, cb](const std::string &topic, const void *data, const int datalen,
void *user_data, const std::string &correlation) -> void {
MSG msg;
msg.SetID(handle);
const AittOption &option);
virtual ~Impl(void);
- void SetWillInfo(const std::string &topic, const void *data, const size_t datalen, AittQoS qos,
+ void SetWillInfo(const std::string &topic, const void *data, const int datalen, AittQoS qos,
bool retain);
void SetConnectionCallback(ConnectionCallback cb, void *user_data);
void Connect(const std::string &host, int port, const std::string &username,
void ConfigureTransportModule(const std::string &key, const std::string &value,
AittProtocol protocols);
- void Publish(const std::string &topic, const void *data, const size_t datalen,
+ void Publish(const std::string &topic, const void *data, const int datalen,
AittProtocol protocols, AittQoS qos, bool retain);
- int PublishWithReply(const std::string &topic, const void *data, const size_t datalen,
+ int PublishWithReply(const std::string &topic, const void *data, const int datalen,
AittProtocol protocol, AittQoS qos, bool retain, const AITT::SubscribeCallback &cb,
void *cbdata, const std::string &correlation);
- int PublishWithReplySync(const std::string &topic, const void *data, const size_t datalen,
+ int PublishWithReplySync(const std::string &topic, const void *data, const int datalen,
AittProtocol protocol, AittQoS qos, bool retain, const SubscribeCallback &cb,
void *cbdata, const std::string &correlation, int timeout_ms);
void ConnectionCB(ConnectionCallback cb, void *user_data, int status);
AittSubscribeID SubscribeMQ(SubscribeInfo *info, MainLoopHandler *loop_handle,
const std::string &topic, const SubscribeCallback &cb, void *cbdata, AittQoS qos);
- void DetachedCB(SubscribeCallback cb, MSG mq_msg, void *data, const size_t datalen,
- void *cbdata, MainLoopHandler::MainLoopResult result, int fd,
- MainLoopHandler::MainLoopData *loop_data);
+ void DetachedCB(SubscribeCallback cb, MSG mq_msg, void *data, const int datalen, void *cbdata,
+ MainLoopHandler::MainLoopResult result, int fd, MainLoopHandler::MainLoopData *loop_data);
void *SubscribeTCP(SubscribeInfo *, const std::string &topic, const SubscribeCallback &cb,
void *cbdata, AittQoS qos);
}
}
-void MosquittoMQ::SetWillInfo(const std::string &topic, const void *msg, size_t szmsg, int qos,
+void MosquittoMQ::SetWillInfo(const std::string &topic, const void *msg, int szmsg, int qos,
bool retain)
{
int ret = mosquitto_will_set(handle, topic.c_str(), szmsg, msg, qos, retain);
subscriber->cb(&mq_msg, msg->topic, msg->payload, msg->payloadlen, subscriber->user_data);
}
-void MosquittoMQ::Publish(const std::string &topic, const void *data, const size_t datalen, int qos,
+void MosquittoMQ::Publish(const std::string &topic, const void *data, const int datalen, int qos,
bool retain)
{
int mid = -1;
}
}
-void MosquittoMQ::PublishWithReply(const std::string &topic, const void *data, const size_t datalen,
+void MosquittoMQ::PublishWithReply(const std::string &topic, const void *data, const int datalen,
int qos, bool retain, const std::string &reply_topic, const std::string &correlation)
{
int ret;
}
}
-void MosquittoMQ::SendReply(MSG *msg, const void *data, const size_t datalen, int qos, bool retain)
+void MosquittoMQ::SendReply(MSG *msg, const void *data, const int datalen, int qos, bool retain)
{
RET_IF(msg == nullptr);
void SetConnectionCallback(const MQConnectionCallback &cb);
void Connect(const std::string &host, int port, const std::string &username,
const std::string &password);
- void SetWillInfo(const std::string &topic, const void *msg, size_t szmsg, int qos, bool retain);
+ void SetWillInfo(const std::string &topic, const void *msg, int szmsg, int qos, bool retain);
void Disconnect(void);
- void Publish(const std::string &topic, const void *data, const size_t datalen, int qos = 0,
+ void Publish(const std::string &topic, const void *data, const int datalen, int qos = 0,
bool retain = false);
- void PublishWithReply(const std::string &topic, const void *data, const size_t datalen, int qos,
+ void PublishWithReply(const std::string &topic, const void *data, const int datalen, int qos,
bool retain, const std::string &reply_topic, const std::string &correlation);
- void SendReply(MSG *msg, const void *data, const size_t datalen, int qos, bool retain);
+ void SendReply(MSG *msg, const void *data, const int datalen, int qos, bool retain);
void *Subscribe(const std::string &topic, const SubscribeCallback &cb,
void *user_data = nullptr, int qos = 0);
void *Unsubscribe(void *handle);
{
}
-void NullTransport::Publish(const std::string& topic, const void* data, const size_t datalen,
+void NullTransport::Publish(const std::string& topic, const void* data, const int datalen,
const std::string& correlation, AittQoS qos, bool retain)
{
}
-void NullTransport::Publish(const std::string& topic, const void* data, const size_t datalen,
+void NullTransport::Publish(const std::string& topic, const void* data, const int datalen,
AittQoS qos, bool retain)
{
}
return nullptr;
}
-void* NullTransport::Subscribe(const std::string& topic, const SubscribeCallback& cb,
- const void* data, const size_t datalen, void* cbdata, AittQoS qos)
-{
- return nullptr;
-}
-
void* NullTransport::Unsubscribe(void* handle)
{
return nullptr;
explicit NullTransport(AittDiscovery &discovery, const std::string &ip);
virtual ~NullTransport(void) = default;
- void Publish(const std::string &topic, const void *data, const size_t datalen,
+ void Publish(const std::string &topic, const void *data, const int datalen,
const std::string &correlation, AittQoS qos = AITT_QOS_AT_MOST_ONCE,
bool retain = false) override;
- void Publish(const std::string &topic, const void *data, const size_t datalen,
+ void Publish(const std::string &topic, const void *data, const int datalen,
AittQoS qos = AITT_QOS_AT_MOST_ONCE, bool retain = false) override;
void *Subscribe(const std::string &topic, const SubscribeCallback &cb, void *cbdata = nullptr,
AittQoS qos = AITT_QOS_AT_MOST_ONCE) override;
- void *Subscribe(const std::string &topic, const SubscribeCallback &cb, const void *data,
- const size_t datalen, void *cbdata = nullptr,
- AittQoS qos = AITT_QOS_AT_MOST_ONCE) override;
-
void *Unsubscribe(void *handle) override;
};
return nullptr;
}
-API int aitt_will_set(aitt_h handle, const char *topic, const void *msg, const size_t msg_len,
+API int aitt_will_set(aitt_h handle, const char *topic, const void *msg, const int msg_len,
aitt_qos_e qos, bool retain)
{
RETV_IF(handle == nullptr, AITT_ERROR_INVALID_PARAMETER);
try {
handle->aitt->SetWillInfo(topic, msg, msg_len, qos, retain);
} catch (std::exception &e) {
- ERR("SetWillInfo(%s, %zu) Fail(%s)", topic, msg_len, e.what());
+ ERR("SetWillInfo(%s, %d) Fail(%s)", topic, msg_len, e.what());
return AITT_ERROR_SYSTEM;
}
return AITT_ERROR_NONE;
return AITT_ERROR_NONE;
}
-API int aitt_publish(aitt_h handle, const char *topic, const void *msg, const size_t msg_len)
+API int aitt_publish(aitt_h handle, const char *topic, const void *msg, const int msg_len)
{
return aitt_publish_full(handle, topic, msg, msg_len, AITT_TYPE_MQTT, AITT_QOS_AT_MOST_ONCE);
}
-API int aitt_publish_full(aitt_h handle, const char *topic, const void *msg, const size_t msg_len,
+API int aitt_publish_full(aitt_h handle, const char *topic, const void *msg, const int msg_len,
int protocols, aitt_qos_e qos)
{
RETV_IF(handle == nullptr, AITT_ERROR_INVALID_PARAMETER);
try {
handle->aitt->Publish(topic, msg, msg_len, AITT_TYPE_MQTT);
} catch (std::exception &e) {
- ERR("Publish(topic:%s, msg_len:%zu) Fail(%s)", topic, msg_len, e.what());
+ ERR("Publish(topic:%s, msg_len:%d) Fail(%s)", topic, msg_len, e.what());
return AITT_ERROR_SYSTEM;
}
}
API int aitt_publish_with_reply(aitt_h handle, const char *topic, const void *msg,
- const size_t msg_len, aitt_protocol_e protocols, aitt_qos_e qos, const char *correlation,
+ const int msg_len, aitt_protocol_e protocols, aitt_qos_e qos, const char *correlation,
aitt_sub_fn cb, void *user_data)
{
RETV_IF(handle == nullptr, AITT_ERROR_INVALID_PARAMETER);
}
API int aitt_send_reply(aitt_h handle, aitt_msg_h msg_handle, const void *reply,
- const size_t reply_len, bool end)
+ const int reply_len, bool end)
{
try {
aitt::MSG *msg = static_cast<aitt::MSG *>(msg_handle);
int cnt = 0;
aitt.Subscribe(
"test/+",
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg,
- void *cbdata) -> void {
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTCPTest *test = static_cast<AITTTCPTest *>(cbdata);
- INFO("Got Message(Topic:%s, size:%zu)", handle->GetTopic().c_str(), szmsg);
+ INFO("Got Message(Topic:%s, size:%d)", handle->GetTopic().c_str(), szmsg);
++cnt;
std::stringstream ss;
aitt.Subscribe(
"test/#",
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTCPTest *test = static_cast<AITTTCPTest *>(cbdata);
- INFO("Got Message(Topic:%s, size:%zu)", handle->GetTopic().c_str(), szmsg);
+ INFO("Got Message(Topic:%s, size:%d)", handle->GetTopic().c_str(), szmsg);
static int cnt = 0;
++cnt;
if (cnt == 3)
aitt.Connect();
aitt.Subscribe(
"test/AITT_will",
- [](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {
+ [](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTManualTest *test = static_cast<AITTManualTest *>(cbdata);
test->ToggleReady();
- DBG("Subscribe invoked: %s %zu", static_cast<const char *>(msg), szmsg);
+ DBG("Subscribe invoked: %s %d", static_cast<const char *>(msg), szmsg);
},
static_cast<void *>(this));
aitt.Connect();
aitt.Subscribe(
testTopic,
- [](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {
+ [](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTest *test = static_cast<AITTTest *>(cbdata);
test->ToggleReady();
if (msg)
- DBG("Subscribe invoked: %s %zu", static_cast<const char *>(msg), szmsg);
+ DBG("Subscribe invoked: %s %d", static_cast<const char *>(msg), szmsg);
else
- DBG("Subscribe invoked: zero size msg(%zu)", szmsg);
+ DBG("Subscribe invoked: zero size msg(%d)", szmsg);
},
static_cast<void *>(this), protocol);
int cnt = 0;
aitt.Subscribe(
TEST_STRESS_TOPIC,
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg,
- void *cbdata) -> void {
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTest *test = static_cast<AITTTest *>(cbdata);
++cnt;
if (szmsg == 0 && cnt != 12) {
int cnt = 0;
aitt.Subscribe(
testTopic,
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg,
- void *cbdata) -> void {
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTest *test = static_cast<AITTTest *>(cbdata);
// NOTE:
// Subscribe callback will be invoked 2 times
++cnt;
const char *receivedMsg = static_cast<const char *>(msg);
- DBG("Subscribe callback called: %d, szmsg = %zu, msg = [%s]", cnt, szmsg,
+ DBG("Subscribe callback called: %d, szmsg = %d, msg = [%s]", cnt, szmsg,
receivedMsg);
if (cnt == 1) {
ASSERT_TRUE(!strcmp(receivedMsg, TEST_MSG));
int cnt = 0;
aitt.Subscribe(
testTopic,
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg,
- void *cbdata) -> void {
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTest *test = static_cast<AITTTest *>(cbdata);
++cnt;
if (cnt == 1)
aitt.Connect();
subscribeHandle = aitt.Subscribe(
testTopic,
- [](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {},
+ [](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {},
nullptr, AITT_TYPE_MQTT);
DBG(">>> Handle: %p", reinterpret_cast<void *>(subscribeHandle));
aitt.Unsubscribe(subscribeHandle);
aitt.Connect();
subscribeHandle = aitt.Subscribe(
testTopic,
- [](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {},
+ [](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {},
nullptr, AITT_TYPE_TCP);
DBG("Subscribe handle: %p", reinterpret_cast<void *>(subscribeHandle));
aitt.Unsubscribe(subscribeHandle);
aitt.Connect();
subscribeHandle = aitt.Subscribe(
testTopic,
- [](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {},
+ [](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {},
nullptr, AITT_TYPE_TCP_SECURE);
DBG("Subscribe handle: %p", reinterpret_cast<void *>(subscribeHandle));
aitt.Unsubscribe(subscribeHandle);
aitt.Connect();
subscribeHandle = aitt.Subscribe(
testTopic,
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTest *test = static_cast<AITTTest *>(cbdata);
- DBG("Subscribe invoked: %s %zu", static_cast<const char *>(msg), szmsg);
+ DBG("Subscribe invoked: %s %d", static_cast<const char *>(msg), szmsg);
static int cnt = 0;
++cnt;
aitt.Connect();
subscribeHandle = aitt.Subscribe(
testTopic,
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {
- DBG("Subscribe invoked: %s %zu", static_cast<const char *>(msg), szmsg);
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
+ DBG("Subscribe invoked: %s %d", static_cast<const char *>(msg), szmsg);
static int cnt = 0;
++cnt;
aitt.Subscribe(
"topic1InCallback",
- [](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) {},
+ [](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) {},
cbdata);
aitt.Subscribe(
"topic2InCallback",
- [](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) {},
+ [](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) {},
cbdata);
DBG("Ready flag is toggled");
g_timeout_add(
aitt.Connect();
aitt.Subscribe(
testTopic,
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTest *test = static_cast<AITTTest *>(cbdata);
- DBG("Subscribe invoked: %s %zu", static_cast<const char *>(msg), szmsg);
+ DBG("Subscribe invoked: %s %d", static_cast<const char *>(msg), szmsg);
test->ToggleReady();
},
static_cast<void *>(this), AITT_TYPE_TCP);
aitt.Subscribe(
testTopic,
- [&](aitt::MSG *handle, const void *msg, const size_t szmsg, void *cbdata) -> void {
+ [&](aitt::MSG *handle, const void *msg, const int szmsg, void *cbdata) -> void {
AITTTest *test = static_cast<AITTTest *>(cbdata);
- DBG("Subscribe invoked: %s %zu", static_cast<const char *>(msg), szmsg);
+ DBG("Subscribe invoked: %s %d", static_cast<const char *>(msg), szmsg);
test->ToggleReady2();
},
static_cast<void *>(this), AITT_TYPE_MQTT);
mq.Connect(LOCAL_IP, 1883, "", "");
mq.Subscribe(
"MQ_TEST_TOPIC1",
- [&](aitt::MSG *handle, const std::string &topic, const void *data,
- const size_t datalen, void *user_data) {
- DBG("Subscribe invoked: %s %zu", static_cast<const char *>(data), datalen);
+ [&](aitt::MSG *handle, const std::string &topic, const void *data, const int datalen,
+ void *user_data) {
+ DBG("Subscribe invoked: %s %d", static_cast<const char *>(data), datalen);
mq.Subscribe(
"topic1InCallback",
[](aitt::MSG *handle, const std::string &topic, const void *msg,
- const size_t szmsg, void *cbdata) {},
+ const int szmsg, void *cbdata) {},
user_data);
mq.Subscribe(
"topic2InCallback",
[](aitt::MSG *handle, const std::string &topic, const void *msg,
- const size_t szmsg, void *cbdata) {},
+ const int szmsg, void *cbdata) {},
user_data);
g_timeout_add(
100,
class AITTRRTest : public testing::Test, public AittTests {
public:
void PublishSyncInCallback(aitt::AITT *aitt, bool *reply1_ok, bool *reply2_ok, aitt::MSG *msg,
- const void *data, const size_t datalen, void *cbdata)
+ const void *data, const int datalen, void *cbdata)
{
aitt->PublishWithReplySync(
rr_topic.c_str(), message.c_str(), message.size(), AITT_TYPE_MQTT,
AITT_QOS_AT_MOST_ONCE, false,
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
CheckReply(msg, data, datalen);
*reply1_ok = true;
},
}
void CheckReplyCallback(bool toggle, bool *reply_ok, aitt::MSG *msg, const void *data,
- const size_t datalen, void *cbdata)
+ const int datalen, void *cbdata)
{
CheckReply(msg, data, datalen);
*reply_ok = true;
void SetUp() override { Init(); }
void TearDown() override { Deinit(); }
- void CheckReply(aitt::MSG *msg, const void *data, const size_t datalen)
+ void CheckReply(aitt::MSG *msg, const void *data, const int datalen)
{
std::string received_data((const char *)data, datalen);
EXPECT_EQ(msg->GetCorrelation(), correlation);
EXPECT_EQ(msg->IsEndSequence(), true);
}
- void CheckSubscribe(aitt::MSG *msg, const void *data, const size_t datalen)
+ void CheckSubscribe(aitt::MSG *msg, const void *data, const int datalen)
{
std::string received_data((const char *)data, datalen);
EXPECT_TRUE(msg->GetTopic() == rr_topic);
aitt.Connect();
aitt.Subscribe(rr_topic.c_str(),
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
CheckSubscribe(msg, data, datalen);
aitt.SendReply(msg, reply.c_str(), reply.size());
sub_ok = true;
INFO("Connected");
sub_aitt.Subscribe(rr_topic.c_str(),
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
CheckSubscribe(msg, data, datalen);
sub_aitt.SendReply(msg, reply.c_str(), reply.size());
sub_ok = true;
aitt.Connect();
aitt.Subscribe(rr_topic.c_str(),
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
CheckSubscribe(msg, data, datalen);
aitt.SendReply(msg, reply.c_str(), reply.size());
sub_ok = true;
aitt.Connect();
aitt.Subscribe(rr_topic.c_str(),
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
CheckSubscribe(msg, data, datalen);
aitt.SendReply(msg, reply1.c_str(), reply1.size(), false);
aitt.PublishWithReply(
rr_topic.c_str(), message.c_str(), message.size(), AITT_TYPE_MQTT,
AITT_QOS_AT_MOST_ONCE, false,
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
std::string reply((const char *)data, datalen);
EXPECT_EQ(msg->GetCorrelation(), correlation);
aitt.Connect();
aitt.Subscribe(rr_topic.c_str(),
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
CheckSubscribe(msg, data, datalen);
aitt.SendReply(msg, reply.c_str(), reply.size());
sub_ok = true;
int ret = aitt.PublishWithReplySync(
rr_topic.c_str(), message.c_str(), message.size(), AITT_TYPE_MQTT,
AITT_QOS_AT_MOST_ONCE, false,
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
FAIL() << "Should not be called";
},
nullptr, correlation, 1);
AITT sub_aitt(clientId + "sub", LOCAL_IP, AittOption(true, false));
sub_aitt.Connect();
sub_aitt.Subscribe(rr_topic.c_str(),
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
INFO("Subscribe Callback is called");
CheckSubscribe(msg, data, datalen);
sub_aitt.SendReply(msg, reply.c_str(), reply.size(), false);
int ret = aitt.PublishWithReplySync(
rr_topic.c_str(), message.c_str(), message.size(), AITT_TYPE_MQTT,
AITT_QOS_AT_MOST_ONCE, false,
- [&](aitt::MSG *msg, const void *data, const size_t datalen, void *cbdata) {
+ [&](aitt::MSG *msg, const void *data, const int datalen, void *cbdata) {
INFO("Reply Callback is called");
static int invalid = 0;
if (invalid)
aitt_sub_h sub_handle = nullptr;
ret = aitt_subscribe(
handle, TEST_C_WILL_TOPIC,
- [](aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data) {
+ [](aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data) {
std::string received_data((const char *)msg, msg_len);
EXPECT_STREQ(received_data.c_str(), TEST_C_MSG);
EXPECT_STREQ(aitt_msg_get_topic(msg_handle), TEST_C_WILL_TOPIC);
aitt_sub_h sub_handle = nullptr;
ret = aitt_subscribe(
handle, TEST_C_TOPIC,
- [](aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data) {
+ [](aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data) {
GMainLoop *loop = static_cast<GMainLoop *>(user_data);
std::string received_data((const char *)msg, msg_len);
EXPECT_STREQ(received_data.c_str(), TEST_C_MSG);
EXPECT_NE(ret, AITT_ERROR_NONE);
ret = aitt_subscribe(
- handle, TEST_C_TOPIC, [](aitt_msg_h, const void *, size_t, void *) {}, nullptr,
- &sub_handle);
+ handle, TEST_C_TOPIC, [](aitt_msg_h, const void *, int, void *) {}, nullptr, &sub_handle);
EXPECT_EQ(ret, AITT_ERROR_NOT_READY);
ret = aitt_subscribe(
- nullptr, TEST_C_TOPIC, [](aitt_msg_h, const void *, size_t, void *) {}, nullptr,
+ nullptr, TEST_C_TOPIC, [](aitt_msg_h, const void *, int, void *) {}, nullptr,
&sub_handle);
EXPECT_EQ(ret, AITT_ERROR_INVALID_PARAMETER);
ASSERT_EQ(ret, AITT_ERROR_NONE);
ret = aitt_subscribe(
- handle, nullptr, [](aitt_msg_h, const void *, size_t, void *) {}, nullptr, &sub_handle);
+ handle, nullptr, [](aitt_msg_h, const void *, int, void *) {}, nullptr, &sub_handle);
EXPECT_EQ(ret, AITT_ERROR_INVALID_PARAMETER);
ret = aitt_subscribe(handle, TEST_C_TOPIC, nullptr, nullptr, &sub_handle);
aitt_sub_h sub_handle = nullptr;
ret = aitt_subscribe(
handle, TEST_C_TOPIC,
- [](aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data) {
+ [](aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data) {
aitt_h handle = static_cast<aitt_h>(user_data);
std::string received_data((const char *)msg, msg_len);
EXPECT_STREQ(received_data.c_str(), TEST_C_MSG);
ret = aitt_publish_with_reply(
handle, TEST_C_TOPIC, TEST_C_MSG, strlen(TEST_C_MSG), AITT_TYPE_MQTT,
AITT_QOS_AT_MOST_ONCE, test_correlation,
- [](aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data) {
+ [](aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data) {
GMainLoop *loop = static_cast<GMainLoop *>(user_data);
std::string received_data((const char *)msg, msg_len);
EXPECT_STREQ(received_data.c_str(), reply_msg);
ret = aitt_publish_with_reply(
nullptr, TEST_C_TOPIC, TEST_C_MSG, strlen(TEST_C_MSG), AITT_TYPE_MQTT,
AITT_QOS_AT_MOST_ONCE, test_correlation,
- [](aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data) {}, nullptr);
+ [](aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data) {}, nullptr);
EXPECT_EQ(ret, AITT_ERROR_INVALID_PARAMETER);
ret = aitt_publish_with_reply(
handle, nullptr, TEST_C_MSG, strlen(TEST_C_MSG), AITT_TYPE_MQTT, AITT_QOS_AT_MOST_ONCE,
test_correlation,
- [](aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data) {}, nullptr);
+ [](aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data) {}, nullptr);
EXPECT_EQ(ret, AITT_ERROR_INVALID_PARAMETER);
ret = aitt_publish_with_reply(
handle, TEST_C_TOPIC, nullptr, 0, AITT_TYPE_MQTT, AITT_QOS_AT_MOST_ONCE, test_correlation,
- [](aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data) {}, nullptr);
+ [](aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data) {}, nullptr);
EXPECT_EQ(ret, AITT_ERROR_INVALID_PARAMETER);
ret = aitt_publish_with_reply(handle, TEST_C_TOPIC, TEST_C_MSG, strlen(TEST_C_MSG),
static aitt_sub_h sub_handle = nullptr;
ret = aitt_subscribe(
handle, TEST_C_TOPIC,
- [](aitt_msg_h msg_handle, const void *msg, size_t msg_len, void *user_data) {
+ [](aitt_msg_h msg_handle, const void *msg, int msg_len, void *user_data) {
sub_call_count++;
},
nullptr, &sub_handle);