tree changes

libs/io_utils/pibroadcast.cpp

@@ -0,0 +1,268 @@
+/*
+	PIP - Platform Independent Primitives
+	Broadcast for all interfaces, including loopback
+	Ivan Pelipenko peri4ko@yandex.ru, Andrey Bychkov work.a.b@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/>.
+*/
+
+#include "pibroadcast.h"
+
+/** \class PIBroadcast
+ * \brief Broadcast for all interfaces, including loopback
+ *
+ * \section PIBroadcast_synopsis Synopsis
+ * %PIBroadcast used as multichannel IO device. It can use
+ * multicast, broadcast and loopback ethernet channels to
+ * send/receive packets. \a send() function send packet to
+ * all initialized ethernets. \a receiveEvent() raised on
+ * packet received by any ethernet. All multi/broadcast
+ * ethernets created for all current addresses, obtained
+ * by \a PIEthernets::allAddresses().
+ *
+ * * \a Multicast ethernets use \a multicastGroup() and \a multicastPort()
+ * * \a Broadcast ethernets use \a broadcastPort()
+ * * \a Loopback ethernet use \a loopbackPortsCount() started from \a loopbackPort()
+ *
+ * %PIBroadcast starts thread, which every 3 seconds check if
+ * current \a PIEthernet::allAddresses() was changed and call
+ * \a reinit() if it necessary.
+ *
+*/
+
+#define MULTICAST_TTL   4
+
+
+PIBroadcast::PIBroadcast(bool send_only): PIThread(), PIEthUtilBase() {
+	_channels = All;
+	eth_lo = 0;
+	mcast_address.set("232.13.3.14", 14100);
+	lo_port = 14200;
+	lo_pcnt = 5;
+	_started = false;
+	_send_only = send_only;
+	_reinit = true;
+	//initMcast(PIEthernet::allAddresses());
+	PIThread::start(3000);
+}
+
+
+PIBroadcast::~PIBroadcast() {
+	PIThread::stop();
+	mcast_mutex.unlock();
+	destroyAll();
+}
+
+
+void PIBroadcast::setChannels(PIBroadcast::Channels ch) {
+	PIMutexLocker ml(mcast_mutex);
+	_channels = ch;
+	_reinit = true;
+}
+
+
+void PIBroadcast::setMulticastGroup(const PIString & mg) {
+	PIMutexLocker ml(mcast_mutex);
+	mcast_address.setIP(mg);
+	_reinit = true;
+}
+
+
+void PIBroadcast::setMulticastPort(ushort port) {
+	PIMutexLocker ml(mcast_mutex);
+	mcast_address.setPort(port);
+	_reinit = true;
+}
+
+
+void PIBroadcast::setMulticastAddress(const PIEthernet::Address & addr) {
+	PIMutexLocker ml(mcast_mutex);
+	mcast_address = addr;
+	_reinit = true;
+}
+
+
+void PIBroadcast::setBroadcastPort(ushort port) {
+	PIMutexLocker ml(mcast_mutex);
+	bcast_port = port;
+	_reinit = true;
+}
+
+
+void PIBroadcast::setLoopbackPort(ushort port) {
+	PIMutexLocker ml(mcast_mutex);
+	lo_port = port;
+	_reinit = true;
+}
+
+
+void PIBroadcast::setLoopbackPortsCount(int count) {
+	PIMutexLocker ml(mcast_mutex);
+	lo_pcnt = count;
+	_reinit = true;
+}
+
+
+void PIBroadcast::destroyAll() {
+	piForeach (PIEthernet * e, eth_mcast) {
+		e->stopThreadedRead();
+		delete e;
+	}
+	eth_mcast.clear();
+	if (eth_lo) {
+		eth_lo->stopThreadedRead();
+		delete eth_lo;
+		eth_lo = 0;
+	}
+}
+
+
+void PIBroadcast::initAll(PIVector<PIEthernet::Address> al) {
+	PIMutexLocker ml(mcast_mutex);
+	destroyAll();
+	_reinit = false;
+	prev_al = al;
+	al.removeAll(PIEthernet::Address("127.0.0.1"));
+	al << mcast_address;
+	eth_mcast.clear();
+	PIEthernet::InterfaceList ifaces = PIEthernet::interfaces();
+	piForeachC (PIEthernet::Address & a, al) {
+		PIEthernet * ce = 0;
+		//piCout << "mcast try" << a;
+
+		if (_channels[Multicast]) {
+			ce = new PIEthernet();
+			ce->setDebug(false);
+			ce->setName("PIMulticast_" + a.toString());
+			ce->setParameters(0);
+			ce->setSendAddress(mcast_address);
+			ce->setMulticastTTL(MULTICAST_TTL);
+			if (!_send_only) {
+				ce->setReadAddress(a.ipString(), mcast_address.port());
+				ce->joinMulticastGroup(mcast_address.ipString());
+				//piCout << "mcast " << ce->readAddress() << ce->sendAddress();
+				if (ce->open()) {
+					eth_mcast << ce;
+					CONNECTU(ce, threadedReadEvent, this, mcastRead);
+				} else {
+					delete ce;
+				}
+			} else {
+				eth_mcast << ce;
+			}
+		}
+
+		if (_channels[Broadcast]) {
+			ce = new PIEthernet();
+			ce->setDebug(false);
+			ce->setName("PIMulticast_" + a.toString());
+			ce->setParameters(PIEthernet::Broadcast);
+			const PIEthernet::Interface * cint = ifaces.getByAddress(a.ipString());
+			PIEthernet::Address nm((cint == 0) ? "255.255.255.0" : cint->netmask);
+			ce->setSendAddress(PIEthernet::getBroadcast(a, nm).ipString(), bcast_port);
+			if (!_send_only) {
+				ce->setReadAddress(PIEthernet::Address(a.ip(), bcast_port));
+				//piCout << "bcast " << ce->readAddress() << ce->sendAddress();
+				if (ce->open()) {
+					eth_mcast << ce;
+					CONNECTU(ce, threadedReadEvent, this, mcastRead);
+				} else {
+					delete ce;
+				}
+			} else {
+				eth_mcast << ce;
+			}
+		}
+
+	}
+
+	if (_channels[Loopback]) {
+		eth_lo = new PIEthernet();
+		eth_lo->setDebug(false);
+		eth_lo->setName("PIMulticast_loopback");
+		if (!_send_only) {
+			eth_lo->setParameter(PIEthernet::ReuseAddress, false);
+			CONNECTU(eth_lo, threadedReadEvent, this, mcastRead);
+			for (int i = 0; i < lo_pcnt; ++i) {
+				eth_lo->setReadAddress("127.0.0.1", lo_port + i);
+				if (eth_lo->open()) {
+					//piCout << "bind local to" << (lo_port + i);
+					break;
+				}
+			}
+		}
+	}
+}
+
+
+void PIBroadcast::send(const PIByteArray & data) {
+	PIByteArray cd = cryptData(data);
+	if (cd.isEmpty()) return;
+	PIMutexLocker ml(mcast_mutex);
+	piForeach (PIEthernet * e, eth_mcast)
+		e->send(cd);
+	if (eth_lo) {
+		for (int i = 0; i < lo_pcnt; ++i) {
+			eth_lo->send("127.0.0.1", lo_port + i, cd);
+		}
+	}
+}
+
+
+void PIBroadcast::startRead() {
+	if (_send_only) return;
+	PIMutexLocker ml(mcast_mutex);
+	piForeach (PIEthernet * e, eth_mcast)
+		e->startThreadedRead();
+	if (eth_lo)
+		eth_lo->startThreadedRead();
+	_started = true;
+}
+
+
+void PIBroadcast::stopRead() {
+	PIMutexLocker ml(mcast_mutex);
+	piForeach (PIEthernet * e, eth_mcast)
+		e->stopThreadedRead();
+	if (eth_lo)
+		eth_lo->stopThreadedRead();
+	_started = false;
+}
+
+
+void PIBroadcast::reinit() {
+	initAll(PIEthernet::allAddresses());
+	if (_started)
+		startRead();
+}
+
+
+void PIBroadcast::mcastRead(uchar * data, int size) {
+	PIByteArray cd = decryptData(PIByteArray(data, size));
+	if (cd.isEmpty()) return;
+	received(cd);
+	receiveEvent(cd);
+}
+
+
+void PIBroadcast::run() {
+	PIVector<PIEthernet::Address> al = PIEthernet::allAddresses();
+	mcast_mutex.lock();
+	bool r = _reinit, ac = (al != prev_al);
+	mcast_mutex.unlock();
+	if (ac || r)
+		reinit();
+	if (ac)
+		addressesChanged();
+}

libs/io_utils/piethutilbase.cpp

@@ -0,0 +1,121 @@
+/*
+	PIP - Platform Independent Primitives
+	Base class for ethernet utils
+	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/>.
+*/
+
+#include "piethutilbase.h"
+#ifdef PIP_CRYPT
+#  include "picrypt.h"
+#endif
+
+/** \class PIEthUtilBase
+ * \brief Base class for ethernet utils
+ *
+ * \section PIEthUtilBase_synopsis Synopsis
+ * %PIEthUtilBase provides crypt layer for derived classes:
+ * \a PIStreamPacker and \a PIBroadcast. All input and output
+ * (sended and received) data can be decrypted/encrypted by this layer.
+ *
+ * By default crypt layer is disabled.
+ *
+ * You can separetely enable it and set ready-to-use
+ * key by \a setCryptEnabled() and \a setCryptKey(). Or you can
+ * use \a createCryptKey() to generate key from your passphrase
+ * and automatic enable crypt layer.
+ *
+ * \note To use crypt layer, PIP should be built with crypt module,
+ * otherwise your in/out data will be lost.
+ *
+ * You can use this class as base for your own classes. Use \a cryptData()
+ * and \a decryptData() when send and receive your data.
+ *
+*/
+
+
+PIEthUtilBase::PIEthUtilBase() {
+	_crypt = false;
+}
+
+
+PIEthUtilBase::~PIEthUtilBase() {
+}
+
+
+void PIEthUtilBase::setCryptEnabled(bool on) {
+	_crypt = on;
+}
+
+
+void PIEthUtilBase::cryptEnable() {
+	setCryptEnabled(true);
+}
+
+
+void PIEthUtilBase::cryptDisable() {
+	setCryptEnabled(false);
+}
+
+
+bool PIEthUtilBase::isCryptEnabled() const {
+	return _crypt;
+}
+
+
+void PIEthUtilBase::setCryptKey(const PIByteArray & k) {
+	_key = k;
+	setCryptEnabled(true);
+}
+
+
+void PIEthUtilBase::createCryptKey(const PIString & k) {
+#ifdef PIP_CRYPT
+	_key = PICrypt::hash("sodium_bug");
+	_key = PICrypt::hash(k);
+#else
+	piCout << "[PIEthUtilBase] PICrypt wasn`t built!";
+#endif
+	_crypt = true;
+}
+
+
+PIByteArray PIEthUtilBase::cryptKey() const {
+	return _key;
+}
+
+
+PIByteArray PIEthUtilBase::cryptData(const PIByteArray & data) {
+	if (!_crypt) return data;
+	return
+#ifdef PIP_CRYPT
+			PICrypt::crypt(data, _key);
+#else
+			PIByteArray();
+#endif
+}
+
+
+PIByteArray PIEthUtilBase::decryptData(const PIByteArray & data) {
+	if (!_crypt) return data;
+#ifdef PIP_CRYPT
+	bool ok = false;
+	PIByteArray ret = PICrypt::decrypt(data, _key, &ok);
+	if (!ok) return PIByteArray();
+	return ret;
+#else
+	return PIByteArray();
+#endif
+}

libs/io_utils/pistreampacker.cpp

@@ -0,0 +1,221 @@
+/*
+	PIP - Platform Independent Primitives
+	Simple packet wrap aroud any PIIODevice
+	Ivan Pelipenko peri4ko@yandex.ru, Andrey Bychkov work.a.b@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/>.
+*/
+
+#ifdef __GNUC__
+#  pragma GCC diagnostic push
+#  pragma GCC diagnostic ignored "-Wnonnull"
+#endif
+#include "pistreampacker.h"
+#include "piiodevice.h"
+#ifdef __GNUC__
+#  pragma GCC diagnostic pop
+#endif
+
+/** \class PIStreamPacker
+ * \brief Simple packet wrap aroud any PIIODevice
+ *
+ * \section PIStreamPacker_synopsis Synopsis
+ * %PIStreamPacker provides simple pack/unpack logic for any data packets.
+ *
+ * When you call \a send() function data splited into several
+ * parts, \a packetSign() prepended to first part and \a sendRequest()
+ * event raised several times.
+ *
+ * When your device receive some data, call \a received() function.
+ * \a packetReceiveEvent() event will be raised when packet will be
+ * collected.
+ *
+ * Use \a assignDevice() to connect device to this %PIStreamPacker.
+ *
+*/
+
+
+PIStreamPacker::PIStreamPacker(PIIODevice * dev): PIObject() {
+	crypt_frag = false;
+	packet_size = -1;
+	crypt_frag_size = 1024*1024;
+	max_packet_size = 1400;
+	packet_sign = 0xAFBE;
+	assignDevice(dev);
+}
+
+
+void PIStreamPacker::send(const PIByteArray & data) {
+	if (data.isEmpty()) return;
+	PIByteArray cd;
+	if (crypt_frag) {
+		int fcnt = (data.size_s() - 1) / crypt_frag_size + 1, fst = 0;
+		//piCout << "crypt_frag send" << fcnt << "frags";
+		PIByteArray frag;
+		for (int i = 0; i < fcnt; ++i) {
+			if (i == fcnt - 1) frag = PIByteArray(data.data(fst), data.size_s() - fst);
+			else frag = PIByteArray(data.data(fst), crypt_frag_size);
+			fst += crypt_frag_size;
+			cd << cryptData(frag);
+		}
+	} else {
+		cd = cryptData(data);
+	}
+	//piCout << "crypt" << data.size() << "->" << cd.size() << key().size();
+	PIByteArray hdr, part;
+	hdr << packet_sign << int(cd.size_s());
+	cd.insert(0, hdr);
+	int pcnt = (cd.size_s() - 1) / max_packet_size + 1, pst = 0;
+	if (pcnt > 1) {
+		prog_s_mutex.lock();
+		prog_s.active = true;
+		prog_s.bytes_all = data.size_s();
+		prog_s.bytes_current = 0;
+		prog_s.progress = 0.;
+		prog_s_mutex.unlock();
+	}
+	for (int i = 0; i < pcnt; ++i) {
+		if (i == pcnt - 1) part = PIByteArray(cd.data(pst), cd.size_s() - pst);
+		else part = PIByteArray(cd.data(pst), max_packet_size);
+		//piCout << "send" << part.size();
+		sendRequest(part);
+		pst += max_packet_size;
+		if (pcnt > 1) {
+			prog_s_mutex.lock();
+			prog_s.bytes_current += part.size_s();
+			prog_s.progress = (double)prog_s.bytes_current / prog_s.bytes_all;
+			prog_s_mutex.unlock();
+		}
+	}
+	if (pcnt > 1) {
+		prog_s_mutex.lock();
+		prog_s.active = false;
+		prog_s_mutex.unlock();
+	}
+}
+
+
+void PIStreamPacker::received(uchar * readed, int size) {
+	received(PIByteArray(readed, size));
+}
+
+
+void PIStreamPacker::received(const PIByteArray & data) {
+	stream.append(data);
+	//piCout << "rec" << data.size();
+	while (stream.size_s() >= 6) {
+		if (packet_size < 0) {
+			ushort sign(0);
+			memcpy(&sign, stream.data(), 2);
+			if (sign != packet_sign) {
+				stream.pop_front();
+				continue;
+			}
+			int sz = -1;
+			memcpy(&sz, stream.data(2), 4);
+			if (sz < 0) {
+				stream.pop_front();
+				continue;
+			}
+			stream.remove(0, 6);
+			packet.clear();
+			packet_size = sz;
+			if (packet_size == 0)
+				packet_size = -1;
+			else {
+				prog_r_mutex.lock();
+				prog_r.active = true;
+				prog_r.bytes_all = packet_size;
+				prog_r.bytes_current = 0;
+				prog_r.progress = 0.;
+				prog_r_mutex.unlock();
+			}
+			continue;
+		} else {
+			int ps = piMini(stream.size_s(), packet_size - packet.size_s());
+			packet.append(stream.data(), ps);
+			prog_r_mutex.lock();
+			prog_r.bytes_current = packet.size_s();
+			prog_r.progress = (double)prog_r.bytes_current / piMaxi(1, prog_r.bytes_all);
+			prog_r_mutex.unlock();
+			stream.remove(0, ps);
+			if (packet.size_s() == packet_size) {
+				PIByteArray cd;
+				if (crypt_frag) {
+					//piCout << "decrypt frags ..." << packet_size;
+					while (packet.size_s() >= 4) {
+						//piCout << "decrypt frags take data ...";
+						PIByteArray frag;
+						//piCout << "decrypt frags take data done" << frag.size_s();
+						packet >> frag;
+						if (frag.isEmpty()) {
+							//piCout << "decrypt frags corrupt, break";
+							cd.clear();
+							break;
+						}
+						cd.append(decryptData(frag));
+						//piCout << "decrypt frags add" << frag.size_s();
+					}
+					//piCout << "decrypt frags done" << cd.size();
+				} else {
+					cd = decryptData(packet);
+				}
+				//piCout << "decrypt" << packet.size() << "->" << cd.size() << key().size();
+				if (!cd.isEmpty()) {
+					packetReceived(cd);
+					packetReceiveEvent(cd);
+				}
+				packet.clear();
+				packet_size = -1;
+				prog_r_mutex.lock();
+				prog_r.active = false;
+				prog_r_mutex.unlock();
+			}
+		}
+	}
+}
+
+
+void PIStreamPacker::assignDevice(PIIODevice * dev) {
+	if (!dev) return;
+	CONNECTU(dev, threadedReadEvent, this, received);
+	CONNECTU(this, sendRequest, dev, write);
+}
+
+
+PIStreamPacker::Progress PIStreamPacker::progressSend() const {
+	PIStreamPacker::Progress ret;
+	prog_s_mutex.lock();
+	ret = prog_s;
+	prog_s_mutex.unlock();
+	return ret;
+}
+
+
+PIStreamPacker::Progress PIStreamPacker::progressReceive() const {
+	PIStreamPacker::Progress ret;
+	prog_r_mutex.lock();
+	ret = prog_r;
+	prog_r_mutex.unlock();
+	return ret;
+}
+
+
+
+
+PIStreamPacker::Progress::Progress() {
+	active = false;
+	bytes_all = bytes_current = 0;
+	progress = 0.;
+}