pip/libs/io_utils/pibroadcast.cpp

/*
    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"

#include "piliterals_time.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;
}


PIBroadcast::~PIBroadcast() {
	PIThread::stopAndWait();
	// 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 PINetworkAddress & 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() {
	for (auto * e: eth_mcast) {
		e->stopAndWait();
		piDeleteSafety(e);
	}
	eth_mcast.clear();
	if (eth_lo) {
		eth_lo->stopAndWait();
		piDeleteSafety(eth_lo);
	}
}


void PIBroadcast::initAll(PIVector<PINetworkAddress> al) {
	PIMutexLocker ml(mcast_mutex);
	destroyAll();
	_reinit = false;
	prev_al = al;
	al.removeAll(PINetworkAddress("127.0.0.1"));
	al << mcast_address;
	eth_mcast.clear();
	PIEthernet::InterfaceList ifaces = PIEthernet::interfaces();
	for (const auto & 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;
					CONNECT2(void, const uchar *, ssize_t, 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());
			PINetworkAddress nm((cint == 0) ? "255.255.255.0" : cint->netmask);
			ce->setSendAddress(PIEthernet::getBroadcast(a, nm).ipString(), bcast_port);
			if (!_send_only) {
				ce->setReadAddress(PINetworkAddress(a.ip(), bcast_port));
				// piCout << "bcast " << ce->readAddress() << ce->sendAddress();
				if (ce->open()) {
					eth_mcast << ce;
					CONNECT2(void, const uchar *, ssize_t, 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);
			CONNECT2(void, const uchar *, ssize_t, 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) {
	/*if (!isRunning()) {
		reinit();
		PIThread::start(3000);
	}*/
	PIByteArray cd = cryptData(data);
	if (cd.isEmpty()) return;
	PIMutexLocker ml(mcast_mutex);
	for (auto * 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 (!isRunning()) {
		_started = false;
		reinit();
		PIThread::start(3_s);
	}
	if (_send_only) return;
	PIMutexLocker ml(mcast_mutex);
	for (auto * e: eth_mcast)
		e->startThreadedRead();
	if (eth_lo) eth_lo->startThreadedRead();
	_started = true;
}


void PIBroadcast::stopRead() {
	if (isRunning()) stopAndWait();
	PIMutexLocker ml(mcast_mutex);
	for (auto * e: eth_mcast)
		e->stopAndWait(5_s);
	if (eth_lo) eth_lo->stopAndWait(5_s);
	_started = false;
}


void PIBroadcast::reinit() {
	initAll(PIEthernet::allAddresses());
	if (_started) startRead();
}


void PIBroadcast::mcastRead(const uchar * data, ssize_t size) {
	PIByteArray cd = decryptData(PIByteArray(data, size));
	if (cd.isEmpty()) return;
	received(cd);
	receiveEvent(cd);
}


void PIBroadcast::run() {
	PIVector<PINetworkAddress> al = PIEthernet::allAddresses();
	mcast_mutex.lock();
	bool r = _reinit, ac = (al != prev_al);
	mcast_mutex.unlock();
	if (ac || r) reinit();
	if (ac) addressesChanged();
}