SHS/pip
4
3
Fork 0
Code Issues 16 Pull Requests 2 Actions Releases Wiki Activity

Merge branch 'master' into pico_sdk

Browse Source
This commit is contained in:
andrey
2026-08-11 09:56:27 +03:00
parent 4e7a5d58aa 9d38731e1a
commit 87c53d45a4
33 changed files with 749 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/client_server/piclientserver_server.cpp
+1
View File
@@ -37,6 +37,7 @@ PIClientServer::Server::Server() {
auto sc = client_factory();
if (!sc) {
piCout << "ClientFactory returns nullptr!"_tr("PIClientServer");
delete c;
return;
}
sc->createForServer(this, c);
libs/console/piterminal.cpp
+1 -1
View File
@@ -880,7 +880,7 @@ bool PITerminal::initialize() {
execvp(argv[0], argv);
delete[] argv[0];
delete[] argv;
exit(0);
exit(errno);
} else {
if (fr < 0 || PRIVATE->fd < 0) {
piCoutObj << "forkpty error," << errorString();
libs/http_server/pihttpserver.cpp
+21 -133
View File
@@ -1,14 +1,25 @@
#include "pihttpserver.h"
#include "piliterals_string.h"
#include "piserverendpoint_p.h"
struct Endpoint: public PIHTTP::ServerEndpoint {
PIHTTP::Method method = PIHTTP::Method::Unknown;
PIHTTPServer::RequestFunction function;
};
PRIVATE_DEFINITION_START(PIHTTPServer)
PIMap<uint, PIVector<Endpoint>> endpoints;
PRIVATE_DEFINITION_END(PIHTTPServer)
PIHTTPServer::PIHTTPServer() {
setRequestCallback([this](const PIHTTP::MessageConst & r) -> PIHTTP::MessageMutable {
PIHTTP::MessageMutable reply;
reply.setCode(PIHTTP::Code::NotFound);
auto in_path = splitPath(r.path());
auto it = endpoints.makeReverseIterator();
auto in_path = PIHTTP::ServerEndpoint::splitPath(r.path());
auto it = PRIVATE->endpoints.makeReverseIterator();
bool found = false;
while (it.next()) {
for (const auto & ep: it.value()) {
@@ -46,7 +57,7 @@ bool PIHTTPServer::registerPath(const PIString & path, PIHTTP::Method method, Re
if (!ep.create(path)) return false;
ep.method = method;
ep.function = std::move(functor);
endpoints[ep.priority] << ep;
PRIVATE->endpoints[ep.priority] << ep;
return true;
}
@@ -57,143 +68,20 @@ void PIHTTPServer::registerUnhandled(RequestFunction functor) {
void PIHTTPServer::unregisterPath(const PIString & path, PIHTTP::Method method) {
auto pl = splitPath(path);
auto it = endpoints.makeIterator();
auto pl = PIHTTP::ServerEndpoint::splitPath(path);
auto it = PRIVATE->endpoints.makeIterator();
while (it.next()) {
it.value().removeWhere([&pl, method](const Endpoint & ep) { return ep.path == pl && ep.method == method; });
}
endpoints.removeWhere([](uint, const PIVector<Endpoint> & epl) { return epl.isEmpty(); });
PRIVATE->endpoints.removeWhere([](uint, const PIVector<Endpoint> & epl) { return epl.isEmpty(); });
}
void PIHTTPServer::unregisterPath(const PIString & path) {
auto pl = splitPath(path);
auto it = endpoints.makeIterator();
auto pl = PIHTTP::ServerEndpoint::splitPath(path);
auto it = PRIVATE->endpoints.makeIterator();
while (it.next()) {
it.value().removeWhere([&pl](const Endpoint & ep) { return ep.path == pl; });
}
endpoints.removeWhere([](uint, const PIVector<Endpoint> & epl) { return epl.isEmpty(); });
}
PIStringList PIHTTPServer::splitPath(const PIString & path) {
auto ret = path.split("/");
ret.removeAll({});
return ret;
}
PIHTTPServer::PathElement::PathElement(const PIString & reg) {
source = reg;
if (reg == "*"_a) {
type = Type::AnyOne;
} else if (reg == "**"_a) {
type = Type::AnyMany;
} else if (reg.contains('*')) {
type = Type::AnyPart;
parts = reg.split('*');
} else if (reg.contains('{')) {
type = Type::Arguments;
int ind = 0, eind = 0, pind = 0;
for (;;) {
ind = reg.find('{', ind);
if (ind < 0) break;
eind = reg.find('}', ind + 1);
if (eind < 0) break;
arguments.insert(arguments.size_s(), reg.mid(ind + 1, eind - ind - 1));
if (ind == 0)
parts << PIString();
else {
if (ind > pind)
parts << reg.mid(pind, ind - pind);
else if (parts.isNotEmpty()) {
piCout << "[PIHTTPServer] Warning: sequential arguments, ignoring this path!";
type = Type::Invalid;
return;
}
}
ind = pind = eind + 1;
}
if (eind < reg.size_s() - 1) parts << reg.mid(eind + 1);
}
}
bool PIHTTPServer::PathElement::match(const PIString & in, PIMap<PIString, PIString> & ext_args) const {
// piCout << "match" << source << "with" << in;
if (type == Type::AnyOne) return true;
if (type == Type::AnyPart) {
int ind = 0;
for (const auto & m: parts) {
ind = in.find(m, ind);
if (ind < 0) return false;
}
return true;
}
if (type == Type::Arguments) {
int ind = 0, eind = 0;
for (int i = 0; i < parts.size_s(); ++i) {
const auto & m(parts[i]);
if (m.isNotEmpty()) {
ind = in.find(m, eind);
if (ind < 0) return false;
}
if (i > 0) {
ext_args[arguments.value(i - 1)] = in.mid(eind, ind - eind);
}
eind = ind + m.size_s();
}
if (parts.size() == arguments.size()) {
ext_args[arguments.value(arguments.size_s() - 1)] = in.mid(eind);
}
return true;
}
return source == in;
}
uint PIHTTPServer::PathElement::priority() const {
switch (type) {
case Type::Fixed: return 0x10000; break;
case Type::Arguments: return 0x1000; break;
case Type::AnyPart: return 0x100; break;
case Type::AnyOne: return 0x10; break;
case Type::AnyMany: return 0x1; break;
default: break;
}
return 0;
}
bool PIHTTPServer::Endpoint::create(const PIString & p) {
path = splitPath(p);
prepared_path.clear();
priority = 0;
for (const auto & i: path) {
PathElement pe(i);
prepared_path << pe;
path_types |= pe.type;
priority += pe.priority();
}
return !path_types[PathElement::Type::Invalid];
}
bool PIHTTPServer::Endpoint::match(const PIStringList & in_path, PIMap<PIString, PIString> & ext_args) const {
if (path_types[PathElement::Type::AnyMany]) {
int any_ind = path.indexOf("**"_a);
for (int i = 0; i < any_ind; ++i) {
if (!prepared_path[i].match(in_path[i], ext_args)) return false;
}
int si = prepared_path.size_s() - 1, ii = in_path.size_s() - 1;
for (; si > any_ind && ii >= 0; --si, --ii) {
if (!prepared_path[si].match(in_path[ii], ext_args)) return false;
}
} else {
if (in_path.size() != prepared_path.size()) return false;
for (int i = 0; i < prepared_path.size_s(); ++i) {
if (!prepared_path[i].match(in_path[i], ext_args)) return false;
}
}
return true;
PRIVATE->endpoints.removeWhere([](uint, const PIVector<Endpoint> & epl) { return epl.isEmpty(); });
}
libs/main/code/picodeparser.cpp
+20 -14
View File
@@ -577,28 +577,36 @@ bool PICodeParser::parseFileContent(PIString & fc, bool main) {
PICodeParser::Entity * PICodeParser::parseClassDeclaration(const PIString & fc) {
static const PIString s_ss = PIStringAscii(" ");
static const PIString s_M = PIStringAscii("$M");
static const PIString s_class = PIStringAscii("class");
PIString cd = fc.trimmed().removeAll('\n').replaceAll('\t', ' ').replaceAll(s_ss, ' '), pn;
static const PIString s_ss = PIStringAscii(" ");
static const PIString s_M = PIStringAscii("$M");
static const PIString s_class = PIStringAscii("class");
static const PIString s_public = PIStringAscii("public");
PIString cd = fc.trimmed().removeAll('\n').replaceAll('\t', ' ').replaceAll(s_ss, ' '), pn;
MetaMap meta;
int ind = cd.find(s_M);
if (ind >= 0) {
meta = tmp_meta.value(cd.takeMid(ind, 5));
cd.replaceAll(s_ss, ' ');
}
PIString typename_ = cd.left(6).trim();
bool is_class = typename_ == s_class;
ind = cd.find(':');
// piCout << "found class <****\n" << cd << "\n****>";
ind = cd.find(':');
PIVector<Entity *> parents;
if (ind > 0) {
PIStringList pl = cd.takeMid(ind + 1).trim().split(',');
cd.cutRight(1);
Entity * pe = 0;
for (const auto & p: pl) {
if (p.contains(' '))
pn = p.mid(p.find(' ') + 1);
else
PIString access;
if (p.contains(' ')) {
access = p.left(p.find(' ')).trim();
pn = p.mid(p.find(' ') + 1);
} else {
pn = p;
}
bool is_public = access.isEmpty() ? !is_class : access == s_public;
if (!is_public) continue;
pe = findEntityByName(pn);
if (pe == 0)
; //{piCout << "Error: can`t find" << pn;}
@@ -606,12 +614,10 @@ PICodeParser::Entity * PICodeParser::parseClassDeclaration(const PIString & fc)
parents << pe;
}
}
PIString typename_ = cd.left(6).trim();
bool is_class = typename_ == s_class;
Visibility vis = cur_def_vis;
cur_def_vis = (is_class ? Private : Public);
PIString cn = cd.mid(6).trim();
bool is_anonymous = cn.isEmpty();
Visibility vis = cur_def_vis;
cur_def_vis = (is_class ? Private : Public);
PIString cn = cd.mid(6).trim();
bool is_anonymous = cn.isEmpty();
if (cn.isEmpty()) cn = PIStringAscii("<unnamed_") + PIString::fromNumber(anon_num++) + '>';
// piCout << "found " << typename_ << cn;
Entity * e = new Entity();
libs/main/core/piwaitevent_p.cpp
+8 -7
View File
@@ -66,9 +66,9 @@ void PIWaitEvent::destroy() {
}
# else
for (int i = 0; i < 2; ++i) {
if (pipe_fd[i] != 0) {
if (pipe_fd[i] != -1) {
::close(pipe_fd[i]);
pipe_fd[i] = 0;
pipe_fd[i] = -1;
}
}
# endif
@@ -90,12 +90,13 @@ bool PIWaitEvent::wait(int fd, CheckRole role) {
FD_SET(pipe_fd[ReadEnd], &(fds[CheckRead]));
FD_SET(fd, &(fds[CheckExeption]));
if (fd_index != CheckExeption) FD_SET(fd, &(fds[fd_index]));
int sr = ::select(nfds, &(fds[CheckRead]), &(fds[CheckWrite]), &(fds[CheckExeption]), nullptr);
int sr = ::select(nfds, &(fds[CheckRead]), &(fds[CheckWrite]), &(fds[CheckExeption]), nullptr);
if (sr < 0) return false;
errorClear();
int buf = 0;
while (::read(pipe_fd[ReadEnd], &buf, sizeof(buf)) > 0)
;
while (::read(pipe_fd[ReadEnd], &buf, sizeof(buf)) > 0) {}
// piCout << "wait result" << sr << FD_ISSET(fd, &(fds[CheckExeption])) << FD_ISSET(fd, &(fds[fd_index]));
if (sr == EBADF || sr == EINTR) return false;
if (errno == EBADF || errno == EINTR) return false;
if (FD_ISSET(fd, &(fds[CheckExeption]))) return true;
return FD_ISSET(fd, &(fds[fd_index]));
# endif
@@ -140,7 +141,7 @@ bool PIWaitEvent::isCreate() const {
# ifdef WINDOWS
return event;
# else
return pipe_fd[ReadEnd] != 0;
return pipe_fd[ReadEnd] != -1;
# endif
}
libs/main/core/piwaitevent_p.h
+1 -1
View File
@@ -57,7 +57,7 @@ private:
# ifdef WINDOWS
void * event = nullptr;
# else
int pipe_fd[2] = {0, 0};
int pipe_fd[2] = {-1, -1};
fd_set fds[3];
enum {
ReadEnd = 0,
libs/main/http_common/piserverendpoint.cpp
+125
View File
@@ -0,0 +1,125 @@
#include "piliterals_string.h"
#include "piserverendpoint_p.h"
PIStringList PIHTTP::ServerEndpoint::splitPath(const PIString & path) {
auto ret = path.split("/");
ret.removeAll({});
return ret;
}
PIHTTP::ServerEndpoint::PathElement::PathElement(const PIString & reg) {
source = reg;
if (reg == "*"_a) {
type = Type::AnyOne;
} else if (reg == "**"_a) {
type = Type::AnyMany;
} else if (reg.contains('*')) {
type = Type::AnyPart;
parts = reg.split('*');
} else if (reg.contains('{')) {
type = Type::Arguments;
int ind = 0, eind = 0, pind = 0;
for (;;) {
ind = reg.find('{', ind);
if (ind < 0) break;
eind = reg.find('}', ind + 1);
if (eind < 0) break;
arguments.insert(arguments.size_s(), reg.mid(ind + 1, eind - ind - 1));
if (ind == 0)
parts << PIString();
else {
if (ind > pind)
parts << reg.mid(pind, ind - pind);
else if (parts.isNotEmpty()) {
piCout << "[PIHTTP::ServerEndpoint] Warning: sequential arguments, ignoring this path!";
type = Type::Invalid;
return;
}
}
ind = pind = eind + 1;
}
if (eind < reg.size_s() - 1) parts << reg.mid(eind + 1);
}
}
bool PIHTTP::ServerEndpoint::PathElement::match(const PIString & in, PIMap<PIString, PIString> & ext_args) const {
// piCout << "match" << source << "with" << in;
if (type == Type::AnyOne) return true;
if (type == Type::AnyPart) {
int ind = 0;
for (const auto & m: parts) {
ind = in.find(m, ind);
if (ind < 0) return false;
}
return true;
}
if (type == Type::Arguments) {
int ind = 0, eind = 0;
for (int i = 0; i < parts.size_s(); ++i) {
const auto & m(parts[i]);
if (m.isNotEmpty()) {
ind = in.find(m, eind);
if (ind < 0) return false;
}
if (i > 0) {
ext_args[arguments.value(i - 1)] = in.mid(eind, ind - eind);
}
eind = ind + m.size_s();
}
if (parts.size() == arguments.size()) {
ext_args[arguments.value(arguments.size_s() - 1)] = in.mid(eind);
}
return true;
}
return source == in;
}
uint PIHTTP::ServerEndpoint::PathElement::priority() const {
switch (type) {
case Type::Fixed: return 0x10000; break;
case Type::Arguments: return 0x1000; break;
case Type::AnyPart: return 0x100; break;
case Type::AnyOne: return 0x10; break;
case Type::AnyMany: return 0x1; break;
default: break;
}
return 0;
}
bool PIHTTP::ServerEndpoint::create(const PIString & p) {
path = splitPath(p);
prepared_path.clear();
priority = 0;
for (const auto & i: path) {
PathElement pe(i);
prepared_path << pe;
path_types |= pe.type;
priority += pe.priority();
}
return !path_types[PathElement::Type::Invalid];
}
bool PIHTTP::ServerEndpoint::match(const PIStringList & in_path, PIMap<PIString, PIString> & ext_args) const {
if (path_types[PathElement::Type::AnyMany]) {
int any_ind = path.indexOf("**"_a);
for (int i = 0; i < any_ind; ++i) {
if (!prepared_path[i].match(in_path[i], ext_args)) return false;
}
int si = prepared_path.size_s() - 1, ii = in_path.size_s() - 1;
for (; si > any_ind && ii >= 0; --si, --ii) {
if (!prepared_path[si].match(in_path[ii], ext_args)) return false;
}
} else {
if (in_path.size() != prepared_path.size()) return false;
for (int i = 0; i < prepared_path.size_s(); ++i) {
if (!prepared_path[i].match(in_path[i], ext_args)) return false;
}
}
return true;
}
libs/main/http_common/piserverendpoint_p.h
+72
View File
@@ -0,0 +1,72 @@
//! \~\file piserverendpoint_p.h
//! \~\ingroup HTTP
//! \~\brief
//! \~english Shared HTTP message container types
//! \~russian Общие типы контейнеров HTTP-сообщений
/*
PIP - Platform Independent Primitives
Shared HTTP message container types
Ivan Pelipenko peri4ko@yandex.ru
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef piserverendpoint_p_h
#define piserverendpoint_p_h
#include "pip_export.h"
#include "pistringlist.h"
namespace PIHTTP {
struct PIP_EXPORT ServerEndpoint {
struct PIP_EXPORT PathElement {
enum class Type {
Invalid = 0x01,
Fixed = 0x02,
Arguments = 0x04,
AnyOne = 0x08,
AnyPart = 0x10,
AnyMany = 0x20
};
Type type = Type::Fixed;
PIString source;
PIStringList parts;
PIMap<int, PIString> arguments;
PathElement(const PIString & reg = {});
bool match(const PIString & in, PIMap<PIString, PIString> & ext_args) const;
uint priority() const;
};
PIStringList path;
PIFlags<PathElement::Type> path_types;
PIVector<PathElement> prepared_path;
uint priority = 0;
bool create(const PIString & p);
bool match(const PIStringList & in_path, PIMap<PIString, PIString> & ext_args) const;
static PIStringList splitPath(const PIString & path);
};
}; // namespace PIHTTP
#endif
libs/main/http_server/pihttpserver.h
+1 -35
View File
@@ -102,43 +102,9 @@ public:
void clearReplyHeaders() { reply_headers.clear(); }
private:
struct PathElement {
enum class Type {
Invalid = 0x01,
Fixed = 0x02,
Arguments = 0x04,
AnyOne = 0x08,
AnyPart = 0x10,
AnyMany = 0x20
};
Type type = Type::Fixed;
PIString source;
PIStringList parts;
PIMap<int, PIString> arguments;
PathElement(const PIString & reg = {});
bool match(const PIString & in, PIMap<PIString, PIString> & ext_args) const;
uint priority() const;
};
struct Endpoint {
PIStringList path;
PIHTTP::Method method = PIHTTP::Method::Unknown;
RequestFunction function;
PIFlags<PathElement::Type> path_types;
PIVector<PathElement> prepared_path;
uint priority = 0;
bool create(const PIString & p);
bool match(const PIStringList & in_path, PIMap<PIString, PIString> & ext_args) const;
};
static PIStringList splitPath(const PIString & path);
PRIVATE_DECLARATION(PIP_HTTP_SERVER_EXPORT)
PIMap<PIString, PIString> reply_headers;
PIMap<uint, PIVector<Endpoint>> endpoints;
RequestFunction unhandled;
};
libs/main/io_devices/pican.cpp
+18 -3
View File
@@ -24,6 +24,7 @@
# define PIP_CAN
#endif
#ifdef PIP_CAN
# include <fcntl.h>
# include <linux/can.h>
# include <linux/can/raw.h>
# include <net/if.h>
@@ -50,7 +51,7 @@ PICAN::PICAN(const PIString & path, PIIODevice::DeviceMode mode): PIIODevice(pat
setPath(path);
#ifdef PIP_CAN
can_id = 0;
sock = 0;
sock = -1;
PRIVATE->event.create();
#endif
}
@@ -71,19 +72,25 @@ bool PICAN::openDevice() {
sock = socket(PF_CAN, SOCK_RAW, CAN_RAW);
if (sock < 0) {
piCoutObj << "Error! while opening socket";
sock = -1;
return false;
}
fcntl(sock, F_SETFL, fcntl(sock, F_GETFL) | O_NONBLOCK);
ifreq ifr;
strcpy(ifr.ifr_name, path().dataAscii());
piCout << "PICAN try to get interface index...";
if (ioctl(sock, SIOCGIFINDEX, &ifr) < 0) {
piCoutObj << "Error! while determin the interface ioctl";
::close(sock);
sock = -1;
return false;
}
struct timeval tv;
tv.tv_sec = 1;
tv.tv_usec = 0;
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char *)&tv, sizeof tv);
if (setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char *)&tv, sizeof tv) < 0) {
piCoutObj << "Error! while setting socket receive timeout";
}
// bind socket to all CAN interface
sockaddr_can addr;
addr.can_family = AF_CAN;
@@ -91,6 +98,8 @@ bool PICAN::openDevice() {
piCout << "PICAN try to bind socket to interface" << ifr.ifr_ifindex;
if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
piCoutObj << "Error! while binding socket";
::close(sock);
sock = -1;
return false;
}
piCout << "PICAN Open OK!";
@@ -105,7 +114,11 @@ bool PICAN::openDevice() {
bool PICAN::closeDevice() {
#ifdef PIP_CAN
interrupt();
if (sock > 0) ::close(sock);
if (sock != -1) {
::shutdown(sock, SHUT_RDWR);
::close(sock);
sock = -1;
}
#endif
return true;
}
@@ -113,6 +126,7 @@ bool PICAN::closeDevice() {
ssize_t PICAN::readDevice(void * read_to, ssize_t max_size) {
#ifdef PIP_CAN
if (sock == -1) return -1;
// piCout << "PICAN read";
can_frame frame;
ssize_t ret = 0;
@@ -131,6 +145,7 @@ ssize_t PICAN::readDevice(void * read_to, ssize_t max_size) {
ssize_t PICAN::writeDevice(const void * data, ssize_t max_size) {
#ifdef PIP_CAN
if (sock == -1) return -1;
// piCout << "PICAN write" << can_id << max_size;
if (max_size > 8) {
piCoutObj << "Can't send CAN frame bigger than 8 bytes (requested " << max_size << ")!";
libs/main/io_devices/pican.h
+1 -1
View File
@@ -73,7 +73,7 @@ protected:
private:
PRIVATE_DECLARATION(PIP_EXPORT)
int sock;
int sock = -1;
int can_id, readed_id;
};
libs/main/io_devices/piethernet.cpp
+20 -6
View File
@@ -103,7 +103,11 @@
# ifndef WINDOWS
PIString getSockAddr(sockaddr * s) {
return s == 0 ? PIString() : PIStringAscii(inet_ntoa(((sockaddr_in *)s)->sin_addr));
if (!s) return PIString();
char buf[INET_ADDRSTRLEN];
piZeroMemory(buf, sizeof(buf));
const char * r = inet_ntop(AF_INET, &((sockaddr_in *)s)->sin_addr, buf, sizeof(buf));
return r ? PIStringAscii(r) : PIString();
}
# endif
@@ -862,6 +866,7 @@ ssize_t PIEthernet::writeDevice(const void * data, ssize_t max_size) {
return -1;
}
}
ret += sr;
remain_data += sr;
remain_size -= sr;
}
@@ -1176,13 +1181,19 @@ PIEthernet::InterfaceList PIEthernet::interfaces() {
# else
# ifdef ANDROID
struct ifconf ifc;
int s = ::socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
int s = ::socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
if (s == -1) {
piCout << "[PIEthernet]"
<< "Can`t create socket: %1"_tr("PIEthernet").arg(errorString());
return il;
}
ifc.ifc_len = 256;
ifc.ifc_buf = new char[ifc.ifc_len];
if (ioctl(s, SIOCGIFCONF, &ifc) < 0) {
piCout << "[PIEthernet]"
<< "Can`t get interfaces: %1"_tr("PIEthernet").arg(errorString());
delete[] ifc.ifc_buf;
::close(s);
return il;
}
int icnt = ifc.ifc_len / sizeof(ifreq);
@@ -1201,7 +1212,8 @@ PIEthernet::InterfaceList PIEthernet::interfaces() {
if (ci.address == "127.0.0.1") ci.flags |= PIEthernet::ifLoopback;
il << ci;
}
delete ifc.ifc_buf;
delete[] ifc.ifc_buf;
::close(s);
# else
struct ifaddrs *ret, *cif = 0;
int s = ::socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
@@ -1223,7 +1235,7 @@ PIEthernet::InterfaceList PIEthernet::interfaces() {
# ifdef QNX
# ifndef BLACKBERRY
int fd = ::open((PIString("/dev/io-net/") + ci.name).dataAscii(), O_RDONLY);
if (fd != 0) {
if (fd >= 0) {
nic_config_t nic;
devctl(fd, DCMD_IO_NET_GET_CONFIG, &nic, sizeof(nic), 0);
::close(fd);
@@ -1292,8 +1304,10 @@ PINetworkAddress PIEthernet::interfaceAddress(const PIString & interface_) {
piZeroMemory(ifr);
strcpy(ifr.ifr_name, interface_.dataAscii());
int s = ::socket(AF_INET, SOCK_DGRAM, 0);
ioctl(s, SIOCGIFADDR, &ifr);
::close(s);
if (s != -1) {
ioctl(s, SIOCGIFADDR, &ifr);
::close(s);
}
struct sockaddr_in * sa = (struct sockaddr_in *)&ifr.ifr_addr;
return PINetworkAddress(uint(sa->sin_addr.s_addr));
# endif
libs/main/io_devices/piiobytearray.cpp
+1 -2
View File
@@ -62,8 +62,7 @@ ssize_t PIIOByteArray::readDevice(void * read_to, ssize_t size) {
if (ret <= 0) return -1;
memcpy(read_to, data_->data(pos), ret);
// piCout << "readed" << ret;
pos += size;
if (pos > data_->size_s()) pos = data_->size_s();
pos += ret;
return ret;
}
libs/main/io_devices/pipeer.cpp
+4 -1
View File
@@ -623,7 +623,10 @@ bool PIPeer::dataRead(const uchar * readed, ssize_t size) {
return true;
}
cnt++;
if (cnt > _PIPEER_MSG_TTL || from == dp->name) return true;
if (cnt > _PIPEER_MSG_TTL || from == dp->name) {
eth_mutex.unlock();
return true;
}
sba << type << from << to << cnt << pba;
// piCout << "translate packet" << from << "->" << to << ", ttl =" << cnt;
sendToNeighbour(dp, sba);
libs/main/io_devices/piserial.cpp
+4 -1
View File
@@ -512,7 +512,10 @@ bool PISerial::read(void * data, int size, double timeout_ms) {
all = readDevice(data, 1);
while (all < size) {
ret = readDevice(&((uchar *)data)[all], size - all);
if (ret > 0) all += ret;
if (ret > 0)
all += ret;
else
break;
}
setOption(BlockingRead, br);
received(data, all);
libs/main/io_utils/pifiletransfer.cpp
+6
View File
@@ -131,6 +131,12 @@ bool PIFileTransfer::sendFiles(const PIVector<PFTFileInfo> & files) {
void PIFileTransfer::processFile(int id, ullong start, PIByteArray & data) {
// piCout << "processFile" << id << files_.size();
if (id <= 0 || id > files_.size_s()) {
cur_file_string = "Error: Invalid file id " + PIString::fromNumber(id);
piCoutObj << cur_file_string;
stopReceive();
return;
}
PFTFileInfo fi = files_[id - 1];
bytes_file_all = fi.size;
bytes_file_cur = start;
libs/main/math/pimathbase.h
+13 -8
View File
@@ -308,10 +308,12 @@ inline PIVector<T> piAbs(const PIVector<T> & v) {
//! \~russian Нормализует угол к диапазону `[0; 360]` градусов на месте.
template<typename T>
void normalizeAngleDeg360(T & a) {
while (a < 0.)
a += 360.;
while (a > 360.)
a -= 360.;
if (std::isnan(a) || std::isinf(a)) {
a = 0.;
return;
}
a -= std::floor(a / 360.) * 360.;
if (a < 0) a += 360;
}
//! \~english Returns an angle normalized to the `[0; 360]` degree range.
@@ -327,10 +329,13 @@ double normalizedAngleDeg360(T a) {
//! \~russian Нормализует угол к диапазону `[-180; 180]` градусов на месте.
template<typename T>
void normalizeAngleDeg180(T & a) {
while (a < -180.)
a += 360.;
while (a > 180.)
a -= 360.;
if (std::isnan(a) || std::isinf(a)) {
a = 0.;
return;
}
a -= std::floor(a / 360.) * 360.;
if (a < -180) a += 360;
if (a >= 180) a -= 360;
}
//! \~english Returns an angle normalized to the `[-180; 180]` degree range.
libs/main/math/pipoint.h
+24
View File
@@ -167,6 +167,22 @@ public:
return *this;
}
//! \~english Multiplies by-coordinates by `v`.
//! \~russian Умножает по-координатно на `v`.
PIPoint<Type> & operator*=(const PIPoint<Type> & v) {
x *= v.x;
y *= v.y;
return *this;
}
//! \~english Divides by-coordinates by `v`.
//! \~russian Делит по-координатно на `v`.
PIPoint<Type> & operator/=(const PIPoint<Type> & v) {
x /= v.x;
y /= v.y;
return *this;
}
//! \~english Returns sum of two points.
//! \~russian Возвращает сумму двух точек.
PIPoint<Type> operator+(const PIPoint<Type> & p) const { return PIPoint<Type>(x + p.x, y + p.y); }
@@ -195,6 +211,14 @@ public:
//! \~russian Возвращает точку, деленную на `v`.
PIPoint<Type> operator/(Type v) const { return PIPoint<Type>(x / v, y / v); }
//! \~english Returns point multiplied by `v`.
//! \~russian Возвращает точку, умноженную на `v`.
PIPoint<Type> operator*(const PIPoint<Type> & v) const { return PIPoint<Type>(x * v.x, y * v.y); }
//! \~english Returns point divided by `v`.
//! \~russian Возвращает точку, деленную на `v`.
PIPoint<Type> operator/(const PIPoint<Type> & v) const { return PIPoint<Type>(x / v.x, y / v.y); }
//! \~english Checks whether point coordinates are equal.
//! \~russian Проверяет равенство координат точек.
bool operator==(const PIPoint<Type> & p) const { return (x == p.x && y == p.y); }
libs/main/mqtt_client/pimqttclient.h
+31 -5
View File
@@ -41,24 +41,39 @@ public:
Client();
virtual ~Client();
//! \~english Request handler used by registered routes and fallback processing.
//! \~russian Обработчик запроса, используемый зарегистрированными маршрутами и fallback-обработкой.
using MessageFunction = std::function<void(const PIMQTT::MessageConst &)>;
void setConnectTimeout(PISystemTime time) { connect_timeout = time; }
void connect(const PIString & address, const PIString & client, const PIString & username = {}, const PIString & password = {});
void disconnect();
void subscribe(const PIString & topic, QoS qos = QoS::Level1);
void subscribe(const PIString & topic, MessageFunction functor, QoS qos = QoS::Level1);
template<typename T>
void
subscribe(const PIString & topic, T * o, PIMQTT::MessageMutable (T::*function)(const PIMQTT::MessageConst &), QoS qos = QoS::Level1) {
subscribe(topic, [o, function](const PIMQTT::MessageConst & m) { return (o->*function)(m); }, qos);
}
void unsubscribe(const PIString & topic);
void unsubscribeAll();
void publish(const PIString & topic, const PIByteArray & msg, QoS qos = QoS::Level0);
void publish(const MessageConst & msg);
void unsubscribeAll() { unsubscribe("#"); }
bool isConnecting() const { return m_status == Connecting; }
bool isConnected() const { return m_status == Connected; }
PIStringList usedTopics() const;
EVENT0(connected);
EVENT1(disconnected, PIMQTT::Error, code);
EVENT1(received, PIMQTT::MessageConst, message);
EVENT1(receivedUnhandled, PIMQTT::MessageConst, message);
struct Endpoint;
private:
NO_COPY_CLASS(Client)
@@ -79,23 +94,34 @@ private:
};
struct Subscribe {
PIString topic;
MessageFunction functor;
QoS qos;
};
void mqtt_connectionLost();
void mqtt_deliveryComplete(int token);
void mqtt_messageArrived(const MessageConst & msg);
void mqtt_messageArrived(MessageMutable & msg);
PIString registerSubscribe(const Subscribe & sub);
PIString unregisterSubscribe(const PIString & mqtt_topic);
void unregisterAll();
void connectInternal(const ConnectInfo & ci);
void disconnectInternal();
void publishInternal(const MessageConst & m);
void subscribeInternal(const Subscribe & sub);
void unsubscribeInternal(const PIString & topic);
void unsubscribeInternal(const PIString & mqtt_topic);
void destroy();
void changeStatus(Status s);
void run();
// from HTTP format
static PIString convertTopic2MQTT(const PIString & topic);
// from MQTT format
static PIString convertTopic2HTTP(const PIString & topic);
std::atomic_int m_status = {Idle};
std::atomic_bool is_destoying = {false};
PISystemTime connect_timeout = 10_s;
libs/main/serialization/pibinarystream.h
+28 -18
View File
@@ -56,17 +56,17 @@
#else
# define BINARY_STREAM_FRIEND(T) \
template<typename P> \
friend PIBinaryStream<P> & operator<<(PIBinaryStream<P> & s, const T & v); \
template<typename P> \
friend PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, T & v);
# define BINARY_STREAM_FRIEND(T) \
template<typename P> \
friend PIBinaryStream<P> & operator<<(PIBinaryStream<P> & s, const T & v); \
template<typename P> \
friend PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, T & v);
# define BINARY_STREAM_WRITE(T) \
template<typename P> \
inline PIBinaryStream<P> & operator<<(PIBinaryStream<P> & s, const T & v)
template<typename P> \
inline PIBinaryStream<P> & operator<<(PIBinaryStream<P> & s, const T & v)
# define BINARY_STREAM_READ(T) \
template<typename P> \
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, T & v)
template<typename P> \
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, T & v)
#endif
@@ -410,7 +410,7 @@ template<typename P,
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIVector<T> & v) {
// piCout << ">> vector trivial default";
int sz = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || sz < 0) {
fprintf(stderr, "error with PIVector<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
@@ -433,7 +433,7 @@ template<typename P,
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIVector<T> & v) {
// piCout << ">> vector trivial custom";
int sz = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || sz < 0) {
fprintf(stderr, "error with PIVector<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
@@ -462,7 +462,7 @@ template<typename P,
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIDeque<T> & v) {
// piCout << ">> deque trivial default";
int sz = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || sz < 0) {
fprintf(stderr, "error with PIDeque<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
@@ -485,7 +485,7 @@ template<typename P,
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIDeque<T> & v) {
// piCout << ">> deque trivial custom";
int sz = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || sz < 0) {
fprintf(stderr, "error with PIDeque<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
@@ -516,7 +516,7 @@ inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIVector2D<T> & v)
int r, c;
r = s.binaryStreamTakeInt();
c = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || r < 0 || c < 0) {
fprintf(stderr, "error with PIVector2D<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
@@ -542,6 +542,11 @@ inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIVector2D<T> & v)
PIVector<T> tmp;
r = s.binaryStreamTakeInt();
c = s.binaryStreamTakeInt();
if (s.wasReadError() || r < 0 || c < 0) {
fprintf(stderr, "error with PIVector2D<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
}
s >> tmp;
if (s.wasReadError()) {
fprintf(stderr, "error with PIVector2D<%s>\n", __PIP_TYPENAME__(T));
@@ -618,7 +623,7 @@ template<typename P, typename T, typename std::enable_if<!std::is_trivially_copy
//! \~russian Восстанавливает %PIVector из нетривиальных элементов по одному.
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIVector<T> & v) {
int sz = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || sz < 0) {
fprintf(stderr, "error with PIVector<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
@@ -641,7 +646,7 @@ template<typename P, typename T, typename std::enable_if<!std::is_trivially_copy
//! \~russian Восстанавливает %PIDeque из нетривиальных элементов по одному.
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIDeque<T> & v) {
int sz = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || sz < 0) {
fprintf(stderr, "error with PIDeque<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
@@ -667,6 +672,11 @@ inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIVector2D<T> & v)
PIVector<T> tmp;
r = s.binaryStreamTakeInt();
c = s.binaryStreamTakeInt();
if (s.wasReadError() || r < 0 || c < 0) {
fprintf(stderr, "error with PIVector2D<%s>\n", __PIP_TYPENAME__(T));
v.clear();
return s;
}
s >> tmp;
if (s.wasReadError()) {
fprintf(stderr, "error with PIVector2D<%s>\n", __PIP_TYPENAME__(T));
@@ -700,7 +710,7 @@ template<typename P, typename Key, typename T>
//! \~russian Восстанавливает ключи и значения %PIMap.
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PIMap<Key, T> & v) {
int sz = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || sz < 0) {
fprintf(stderr, "error with PIMap<%s, %s>\n", __PIP_TYPENAME__(Key), __PIP_TYPENAME__(T));
v.clear();
return s;
@@ -749,7 +759,7 @@ template<typename P, typename Key>
//! \~russian Восстанавливает ключи %PISet.
inline PIBinaryStream<P> & operator>>(PIBinaryStream<P> & s, PISet<Key> & v) {
int sz = s.binaryStreamTakeInt();
if (s.wasReadError()) {
if (s.wasReadError() || sz < 0) {
fprintf(stderr, "error with PISet<%s>\n", __PIP_TYPENAME__(Key));
v.clear();
return s;
libs/main/serialization/pijson.cpp
+1 -1
View File
@@ -547,7 +547,7 @@ void PIJSON::print(PIString & s, const PIJSON & v, PIString tab, bool spaces, bo
if (spaces) s += ' ';
}
switch (v.c_type) {
case PIJSON::Invalid: break;
case PIJSON::Invalid:
case PIJSON::Null: s += "null"; break;
case PIJSON::Boolean: s += PIString::fromBool(v.c_value.toBool()); break;
case PIJSON::Number: s += v.c_value.toString(); break;
libs/main/serialization/pijsonserialization.h
+14
View File
@@ -291,6 +291,13 @@ inline PIJSON piSerializeJSON(const PIMathVectorT<Size, T> & v) {
return ret;
}
template<>
//! \~english Serializes %PIStringList as JSON array.
//! \~russian Сериализует %PIStringList как JSON-массив.
inline PIJSON piSerializeJSON(const PIStringList & v) {
return piSerializeJSON((const PIDeque<PIString> &)v);
}
// ---
// deserialize, piDeserializeJSON(T, PIJSON)
@@ -538,6 +545,13 @@ inline void piDeserializeJSON(PIMathVectorT<Size, T> & v, const PIJSON & js) {
piDeserializeJSON(v[i], js[i]);
}
template<>
//! \~english Deserializes %PIStringList from JSON array.
//! \~russian Десериализует %PIStringList из JSON-массива.
inline void piDeserializeJSON(PIStringList & v, const PIJSON & js) {
piDeserializeJSON((PIDeque<PIString> &)v, js);
}
// ---
// PIJSON static wrapper
libs/main/system/piprocess.cpp
+14 -7
View File
@@ -215,10 +215,10 @@ PRIVATE_DEFINITION_START(PIProcess)
PeekNamedPipe(pipes[pipe_type][PipeRead], nullptr, 0, nullptr, &available, nullptr);
if (available > 0) {
BOOL ok = ReadFile(pipes[pipe_type][PipeRead],
read_buffer.data(offset),
piMini(available, read_buffer.size() - offset),
&bytes_read,
nullptr);
read_buffer.data(offset),
piMini(available, read_buffer.size() - offset),
&bytes_read,
nullptr);
if (!ok) bytes_read = 0;
}
# else
@@ -293,10 +293,10 @@ void PIProcess::startProc(bool detached) {
si.dwFlags |= STARTF_USESTDHANDLES;
const auto cmd = convertWindowsCmd(args);
if (CreateProcessA(0, // No module name (use command line)
(LPSTR)cmd.data(), // Command line
(LPSTR)cmd.data(), // Command line
0, // Process handle not inheritable
0, // Thread handle not inheritable
true, // Set handle inheritance to FALSE
true, // Set handle inheritance to FALSE
detached ? DETACHED_PROCESS /*CREATE_NEW_CONSOLE*/ : 0, // Creation flags
0, // Use environment
wd.isEmpty() ? 0 : wd.data(), // Use working directory
@@ -319,10 +319,17 @@ void PIProcess::startProc(bool detached) {
auto largs = convertToCharArrays(args);
auto lenv = convertToCharArrays(env);
int pid_ = fork();
if (pid_ < 0) {
piCoutObj << "\"fork\" error: " << errorString();
PRIVATE->closeAllPipes();
delete[] largs;
delete[] lenv;
return;
}
if (!detached) PRIVATE->pid = pid_;
if (pid_ == 0) {
if (!wd.isEmpty()) {
if (!chdir(wd.data())) piCoutObj << "Error while set working directory";
if (chdir(wd.data()) != 0) piCoutObj << "Error while set working directory";
}
PRIVATE->closePipe(StdIn, PipeWrite);
PRIVATE->closePipe(StdOut, PipeRead);
libs/main/text/pistring.cpp
+2 -2
View File
@@ -671,7 +671,7 @@ PIString & PIString::operator+=(const PIConstChars & str) {
if (!str.isEmpty()) {
size_t os = d.size();
d.enlarge(str.size());
for (size_t l = 0; l < d.size(); ++l) {
for (size_t l = 0; l < str.size(); ++l) {
d[os + l] = str[l];
}
}
@@ -1763,7 +1763,7 @@ PIString PIString::toLowerCase() const {
char PIString::toChar() const {
char v;
char v = 0;
sscanf(dataAscii(), "%c", &v);
return v;
}
libs/main/types/pibytearray.h
+4 -2
View File
@@ -1107,9 +1107,11 @@ public:
PIByteArray & append(const PIByteArray & data_) {
#ifdef CC_GCC
# pragma GCC diagnostic push
# pragma GCC diagnostic ignored "-Wstringop-overflow"
# pragma GCC diagnostic ignored "-Warray-bounds"
# pragma GCC diagnostic ignored "-Wrestrict"
# ifndef ANDROID
# pragma GCC diagnostic ignored "-Wstringop-overflow"
# pragma GCC diagnostic ignored "-Wrestrict"
# endif
#endif
const size_t ps = size();
enlarge(data_.size_s());
libs/main/types/pivaluetree.cpp
+1 -1
View File
@@ -217,7 +217,7 @@ PIValueTree & PIValueTree::child(const PIStringList & path) {
if (_is_null || path.isEmpty()) return *this;
PIValueTree * ret = &child(path[0]);
for (int i = 1; i < path.size_s(); ++i)
ret = &child(path[i]);
ret = &(ret->child(path[i]));
return *ret;
}
libs/mqtt_client/pimqttclient.cpp
+193 -11
View File
@@ -20,6 +20,36 @@
#include "pimqttclient.h"
#include "MQTTClient.h"
#include "piliterals_string.h"
#include "piserverendpoint_p.h"
struct PIMQTT::Client::Endpoint: public PIHTTP::ServerEndpoint {
PIMQTT::Client::MessageFunction function;
};
struct TopicUsage {
int counter = 0;
PIMQTT::QoS qos = PIMQTT::QoS::Level1;
};
struct EndpointsStorage {
PIMap<uint, PIMap<PIString, PIVector<PIMQTT::Client::Endpoint>>> prepared; // [priority][topic] -> endpoints
PIMap<PIString, TopicUsage> topics_binded; // [topic] -> TopicUsage
};
PIString topicFromEndpoint(const PIMQTT::Client::Endpoint & e) {
PIStringList ret;
for (const auto & i: e.prepared_path) {
switch (i.type) {
case PIHTTP::ServerEndpoint::PathElement::Type::Fixed: ret << i.source; break;
case PIHTTP::ServerEndpoint::PathElement::Type::AnyMany: ret << "#"_a; break;
default: ret << "+"_a; break;
}
}
return ret.join('/');
}
STATIC_INITIALIZER_BEGIN
@@ -30,7 +60,9 @@ STATIC_INITIALIZER_END
PRIVATE_DEFINITION_START(PIMQTT::Client)
MQTTClient client = nullptr;
bool connected = false;
std::atomic<bool> connected{false};
PIProtectedVariable<EndpointsStorage> endpoints;
static void connectionLost_callback(void * context, char *) {
((PIMQTT::Client *)context)->mqtt_connectionLost();
@@ -88,15 +120,23 @@ void PIMQTT::Client::disconnect() {
}
void PIMQTT::Client::subscribe(const PIString & topic, QoS qos) {
void PIMQTT::Client::subscribe(const PIString & topic, MessageFunction functor, QoS qos) {
if (is_destoying) return;
worker->enqueueTask([this, topic, qos] { subscribeInternal({topic, qos}); });
Subscribe sub{topic, functor, qos};
// piCout << "subscribe" << topic;
PIString mqtt_topic = registerSubscribe(sub);
if (mqtt_topic.isEmpty()) return;
sub.topic = mqtt_topic;
worker->enqueueTask([this, sub] { subscribeInternal(sub); });
}
void PIMQTT::Client::unsubscribe(const PIString & topic) {
if (is_destoying) return;
worker->enqueueTask([this, topic] { unsubscribeInternal(topic); });
// piCout << "unsubscribe" << topic;
PIString mqtt_topic = unregisterSubscribe(topic);
if (mqtt_topic.isEmpty()) return;
worker->enqueueTask([this, mqtt_topic] { unsubscribeInternal(mqtt_topic); });
}
@@ -115,8 +155,27 @@ void PIMQTT::Client::publish(const MessageConst & msg) {
}
PIStringList PIMQTT::Client::usedTopics() const {
return PRIVATE->endpoints.getRef()->topics_binded.keys();
}
void PIMQTT::Client::unsubscribeAll() {
{
auto ref = PRIVATE->endpoints.getRef();
auto tit = ref->topics_binded.makeIterator();
while (tit.next()) {
if (tit.value().counter <= 0) continue;
PIString mqtt_topic = tit.key();
worker->enqueueTask([this, mqtt_topic] { unsubscribeInternal(mqtt_topic); });
}
}
unregisterAll();
}
void PIMQTT::Client::mqtt_connectionLost() {
piCoutObj << "mqtt_connectionLost";
// piCoutObj << "mqtt_connectionLost";
PRIVATE->connected = false;
changeStatus(Idle);
disconnected(Error::ServerUnavailable);
@@ -126,9 +185,114 @@ void PIMQTT::Client::mqtt_connectionLost() {
void PIMQTT::Client::mqtt_deliveryComplete(int token) {}
void PIMQTT::Client::mqtt_messageArrived(const MessageConst & msg) {
piCoutObj << "mqtt_messageArrived";
received(msg);
void PIMQTT::Client::mqtt_messageArrived(MessageMutable & msg) {
PIStringList in_path = msg.topicList();
PIMQTT::Client::MessageFunction function;
// piCoutObj << "mqtt_messageArrived";
{
bool found = false;
auto ref = PRIVATE->endpoints.getRef();
auto pit = ref->prepared.makeReverseIterator();
while (pit.next()) { // by priority
auto tit = pit.value().makeIterator();
while (tit.next()) { // by MQTT topic
for (const auto & ep: tit.value()) {
PIMap<PIString, PIString> ext_args;
if (ep.match(in_path, ext_args)) {
msg.pathArguments() = ext_args;
function = ep.function;
found = true;
break;
}
}
if (found) break;
}
if (found) break;
}
}
if (function)
function(msg);
else
receivedUnhandled(msg);
}
PIString PIMQTT::Client::registerSubscribe(const Subscribe & sub) {
Endpoint ep;
ep.create(convertTopic2HTTP(sub.topic));
ep.function = sub.functor;
PIString topic = topicFromEndpoint(ep);
if (topic.isEmpty()) {
piCoutObj << "Warning: subscribe to empty topic, ignore";
return {};
}
// piCout << sub.topic << "->" << topic << ep.priority;
bool is_new_topic = false;
auto ref = PRIVATE->endpoints.getRef();
auto & eps_by_topic(ref->prepared[ep.priority][topic]);
for (const auto & i: eps_by_topic) {
if (i.path == ep.path) {
piCoutObj << "Warning: subscribe duplicate path, ignore";
return {};
}
}
eps_by_topic << ep;
auto & usage(ref->topics_binded[topic]);
is_new_topic = usage.counter == 0;
++usage.counter;
if (!is_new_topic) return {};
usage.qos = sub.qos;
return topic;
}
PIString PIMQTT::Client::unregisterSubscribe(const PIString & mqtt_topic) {
Endpoint ep;
ep.create(convertTopic2HTTP(mqtt_topic));
PIString topic = topicFromEndpoint(ep);
if (topic.isEmpty()) {
piCoutObj << "Warning: unsubscribe from empty topic, ignore";
return {};
}
// piCout << mqtt_topic << "->" << topic << ep.priority;
auto ref = PRIVATE->endpoints.getRef();
auto pit = ref->prepared.makeIterator();
while (pit.next()) { // by priority
auto tit = pit.value().makeIterator();
while (tit.next()) { // by MQTT topic
auto & eps(tit.value());
for (int i = 0; i < eps.size_s(); ++i) {
if (eps[i].path == ep.path) {
eps.remove(i);
auto & usage(ref->topics_binded[tit.key()]);
--usage.counter;
PIString ret;
if (usage.counter <= 0) {
ret = tit.key();
ref->topics_binded.remove(tit.key());
}
if (eps.isEmpty()) {
// piCout << "remove topics" << tit.key();
pit.value().remove(tit.key());
}
return ret;
}
}
}
}
piCoutObj << "Warning: unsubscribe from" << mqtt_topic << ", topic not found";
return {};
}
void PIMQTT::Client::unregisterAll() {
auto ref = PRIVATE->endpoints.getRef();
ref->prepared.clear();
ref->topics_binded.clear();
}
@@ -160,6 +324,12 @@ void PIMQTT::Client::connectInternal(const ConnectInfo & ci) {
return;
}
PRIVATE->connected = true;
PIMap<PIString, TopicUsage> topics_binded;
{ topics_binded = PRIVATE->endpoints.getRef()->topics_binded; }
auto it = topics_binded.makeIterator();
while (it.next()) {
if (it.value().counter > 0) subscribeInternal({it.key(), nullptr, it.value().qos});
}
changeStatus(Connected);
connected();
}
@@ -189,18 +359,21 @@ void PIMQTT::Client::publishInternal(const MessageConst & m) {
void PIMQTT::Client::subscribeInternal(const Subscribe & sub) {
if (!PRIVATE->client) return;
// piCout << "subscribeInternal" << sub.topic;
int ret = MQTTClient_subscribe(PRIVATE->client, sub.topic.dataUTF8(), static_cast<int>(sub.qos));
if (ret != MQTTCLIENT_SUCCESS) {
piCoutObj << "Failed to subscribe" << sub.topic << ", code" << ret;
return;
}
}
void PIMQTT::Client::unsubscribeInternal(const PIString & topic) {
void PIMQTT::Client::unsubscribeInternal(const PIString & mqtt_topic) {
if (!PRIVATE->client) return;
int ret = MQTTClient_unsubscribe(PRIVATE->client, topic.dataUTF8());
// piCout << "unsubscribeInternal" << mqtt_topic;
int ret = MQTTClient_unsubscribe(PRIVATE->client, mqtt_topic.dataUTF8());
if (ret != MQTTCLIENT_SUCCESS) {
piCoutObj << "Failed to unsubscribe" << topic << ", code" << ret;
piCoutObj << "Failed to unsubscribe" << mqtt_topic << ", code" << ret;
}
}
@@ -217,3 +390,12 @@ void PIMQTT::Client::destroy() {
void PIMQTT::Client::changeStatus(Status s) {
m_status = s;
}
PIString PIMQTT::Client::convertTopic2MQTT(const PIString & topic) {
return topic.replacedAll("**", '#').replacedAll('*', '+');
}
PIString PIMQTT::Client::convertTopic2HTTP(const PIString & topic) {
return topic.replacedAll('#', "**").replacedAll('+', '*');
}