pip/src_io_utils/pibroadcast.cpp

/*
	PIP - Platform Independent Primitives
	Broadcast for all interfaces, including loopback
	Copyright (C) 2019  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 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 General Public License for more details.

	You should have received a copy of the GNU 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();
}