syn2cat
/
syndilights
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
syndilights/open-lighting-architecture/ola-0.8.4/plugins/artnet/ArtNetNode.cpp

1629 lines
50 KiB
C++
Raw Normal View History

added Open Lighting Architecture files
2010-10-31 17:32:25 +01:00
/*
* 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 2 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 Library General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
*
* ArtNetNode.cpp
* An ArtNet node
* Copyright (C) 2005-2010 Simon Newton
*/
#include <string.h>
#include <algorithm>
#include <map>
#include <set>
#include <string>
#include "ola/BaseTypes.h"
#include "ola/Logging.h"
#include "ola/network/NetworkUtils.h"
#include "plugins/artnet/ArtNetNode.h"
namespace ola {
namespace plugin {
namespace artnet {
using ola::Callback1;
using ola::Closure;
using ola::network::AddressToString;
using ola::network::HostToLittleEndian;
using ola::network::HostToNetwork;
using ola::network::LittleEndianToHost;
using ola::network::NetworkToHost;
using ola::network::UdpSocket;
using std::pair;
using std::string;
const char ArtNetNode::ARTNET_ID[] = "Art-Net";
/*
* Create a new node
* @param interface the interface to use.
* @param short_name the short node name
* @param long_name the long node name
* @param subnet_address the ArtNet 'subnet' address, 4 bits.
*/
ArtNetNode::ArtNetNode(const ola::network::Interface &interface,
const string &short_name,
const string &long_name,
const PluginAdaptor *plugin_adaptor,
uint8_t subnet_address)
: m_running(false),
m_send_reply_on_change(true),
m_short_name(short_name),
m_long_name(long_name),
m_broadcast_threshold(BROADCAST_THRESHOLD),
m_unsolicited_replies(0),
m_plugin_adaptor(plugin_adaptor),
m_interface(interface),
m_socket(NULL),
m_discovery_timeout(ola::network::INVALID_TIMEOUT) {
// reset all the port structures
for (unsigned int i = 0; i < ARTNET_MAX_PORTS; i++) {
m_input_ports[i].universe_address = 0;
m_input_ports[i].sequence_number = 0;
m_input_ports[i].enabled = false;
m_input_ports[i].on_tod = NULL;
m_input_ports[i].on_rdm_response = NULL;
m_input_ports[i].discovery_running = false;
m_input_ports[i].rdm_send_timeout = ola::network::INVALID_TIMEOUT;
m_input_ports[i].overflowed_response = NULL;
m_output_ports[i].universe_address = 0;
m_output_ports[i].sequence_number = 0;
m_output_ports[i].enabled = false;
m_output_ports[i].is_merging = false;
m_output_ports[i].merge_mode = ARTNET_MERGE_HTP;
m_output_ports[i].buffer = NULL;
m_output_ports[i].on_data = NULL;
m_output_ports[i].on_discover = NULL;
m_output_ports[i].on_flush = NULL;
m_output_ports[i].on_rdm_request = NULL;
for (unsigned int j = 0; j < MAX_MERGE_SOURCES; j++)
m_output_ports[i].sources[j].address.s_addr = 0;
}
SetSubnetAddress(subnet_address);
}
/*
* Cleanup
*/
ArtNetNode::~ArtNetNode() {
Stop();
for (unsigned int i = 0; i < ARTNET_MAX_PORTS; i++) {
if (m_input_ports[i].on_tod)
delete m_input_ports[i].on_tod;
if (m_input_ports[i].on_rdm_response)
delete m_input_ports[i].on_rdm_response;
if (m_output_ports[i].on_data)
delete m_output_ports[i].on_data;
if (m_output_ports[i].on_discover)
delete m_output_ports[i].on_discover;
if (m_output_ports[i].on_flush)
delete m_output_ports[i].on_flush;
if (m_output_ports[i].on_rdm_request)
delete m_output_ports[i].on_rdm_request;
}
}
/*
* Start this node. The port modifying functions can be called before this.
*/
bool ArtNetNode::Start() {
if (m_running)
return false;
if (!InitNetwork())
return false;
m_running = true;
MaybeSendPoll(); // send a poll, this will result in us repling as well
SendPollReply(m_interface.bcast_address);
return true;
}
/*
* Stop this node
*/
bool ArtNetNode::Stop() {
if (!m_running)
return false;
if (m_discovery_timeout != ola::network::INVALID_TIMEOUT) {
m_plugin_adaptor->RemoveTimeout(m_discovery_timeout);
m_discovery_timeout = ola::network::INVALID_TIMEOUT;
}
for (unsigned int i = 0; i < ARTNET_MAX_PORTS; i++) {
if (m_input_ports[i].rdm_send_timeout != ola::network::INVALID_TIMEOUT) {
m_plugin_adaptor->RemoveTimeout(m_input_ports[i].rdm_send_timeout);
m_input_ports[i].rdm_send_timeout = ola::network::INVALID_TIMEOUT;
}
// clean up any pending RDM messsages.
while (!m_input_ports[i].pending_rdm_requests.empty()) {
delete m_input_ports[i].pending_rdm_requests.front();
m_input_ports[i].pending_rdm_requests.pop();
}
}
m_plugin_adaptor->RemoveSocket(m_socket);
if (m_socket) {
delete m_socket;
m_socket = NULL;
}
m_running = false;
return true;
}
/*
* Set the short name
*/
bool ArtNetNode::SetShortName(const string &name) {
m_short_name = name;
if (m_running && m_send_reply_on_change) {
m_unsolicited_replies++;
return SendPollReply(m_interface.bcast_address);
}
return true;
}
/*
* Set the long name
*/
bool ArtNetNode::SetLongName(const string &name) {
m_long_name = name;
if (m_running && m_send_reply_on_change) {
m_unsolicited_replies++;
return SendPollReply(m_interface.bcast_address);
}
return true;
}
/*
* The the subnet address for this node
*/
bool ArtNetNode::SetSubnetAddress(uint8_t subnet_address) {
uint8_t old_address = m_input_ports[0].universe_address >> 4;
if (old_address == subnet_address)
return true;
subnet_address = subnet_address << 4;
for (unsigned int i = 0; i < ARTNET_MAX_PORTS; i++) {
m_input_ports[i].universe_address = subnet_address |
(m_input_ports[i].universe_address & 0x0f);
m_output_ports[i].universe_address = subnet_address |
(m_output_ports[i].universe_address & 0x0f);
m_input_ports[i].uids.clear();
}
if (m_running && m_send_reply_on_change) {
m_unsolicited_replies++;
return SendPollReply(m_interface.bcast_address);
}
return true;
}
/*
* Set the universe for a port
*/
bool ArtNetNode::SetPortUniverse(artnet_port_type type,
uint8_t port_id,
uint8_t universe_id) {
if (port_id > ARTNET_MAX_PORTS) {
OLA_WARN << "Port index of out bounds: " << port_id << " > " <<
ARTNET_MAX_PORTS;
return false;
}
if (type == ARTNET_INPUT_PORT) {
uint8_t old_universe = m_input_ports[port_id].universe_address;
m_input_ports[port_id].universe_address = (
(universe_id & 0x0f) |
(m_input_ports[port_id].universe_address & 0xf0));
if (old_universe != m_input_ports[port_id].universe_address)
// clear the uid map
m_input_ports[port_id].uids.clear();
bool ports_previously_enabled = false;
for (unsigned int i = 0; i < ARTNET_MAX_PORTS; i++)
ports_previously_enabled |= m_input_ports[i].enabled;
m_input_ports[port_id].enabled = universe_id != ARTNET_DISABLE_PORT;
if (!ports_previously_enabled && m_input_ports[port_id].enabled)
MaybeSendPoll();
} else {
m_output_ports[port_id].universe_address = (
(universe_id & 0x0f) |
(m_output_ports[port_id].universe_address & 0xf0));
m_output_ports[port_id].enabled = universe_id != ARTNET_DISABLE_PORT;
}
if (m_running && m_send_reply_on_change) {
m_unsolicited_replies++;
return SendPollReply(m_interface.bcast_address);
}
return true;
}
/*
* Return the current universe address for a port
*/
uint8_t ArtNetNode::GetPortUniverse(artnet_port_type type, uint8_t port_id) {
if (port_id > ARTNET_MAX_PORTS) {
OLA_WARN << "Port index of out bounds: " << port_id << " > " <<
ARTNET_MAX_PORTS;
return 0;
}
if (type == ARTNET_INPUT_PORT)
return m_input_ports[port_id].universe_address;
else
return m_output_ports[port_id].universe_address;
}
/*
* Set the merge mode for an output port
*/
bool ArtNetNode::SetMergeMode(uint8_t port_id, artnet_merge_mode merge_mode) {
if (port_id > ARTNET_MAX_PORTS) {
OLA_WARN << "Port index of out bounds: " << port_id << " > " <<
ARTNET_MAX_PORTS;
return false;
}
m_output_ports[port_id].merge_mode = merge_mode;
if (m_running && m_send_reply_on_change) {
m_unsolicited_replies++;
return SendPollReply(m_interface.bcast_address);
}
return true;
}
/*
* Send an ArtPoll if any of the ports are sending data
*/
bool ArtNetNode::MaybeSendPoll() {
if (!m_running)
return false;
bool send = false;
for (unsigned int i = 0; i < ARTNET_MAX_PORTS; i++)
send |= m_input_ports[i].enabled;
if (!send)
return true;
OLA_DEBUG << "Sending ArtPoll";
artnet_packet packet;
PopulatePacketHeader(&packet, ARTNET_POLL);
memset(&packet.data.poll, 0, sizeof(packet.data.poll));
packet.data.poll.version = HostToNetwork(ARTNET_VERSION);
// send PollReplies when something changes
packet.data.poll.talk_to_me = 0x02;
unsigned int size = sizeof(packet.data.poll);
return SendPacket(packet, size, m_interface.bcast_address);
}
/*
* Send some DMX data
* @param universe the id of the universe to send
* @param buffer the DMX data
* @return true if it was send successfully, false otherwise
*/
bool ArtNetNode::SendDMX(uint8_t port_id, const DmxBuffer &buffer) {
if (!CheckInputPortState(port_id, "ArtDMX"))
return false;
if (!buffer.Size()) {
OLA_DEBUG << "Not sending 0 length packet";
return true;
}
artnet_packet packet;
PopulatePacketHeader(&packet, ARTNET_DMX);
memset(&packet.data.dmx, 0, sizeof(packet.data.dmx));
packet.data.poll.version = HostToNetwork(ARTNET_VERSION);
packet.data.dmx.sequence = m_input_ports[port_id].sequence_number;
packet.data.dmx.physical = 1 + port_id;
packet.data.dmx.universe = m_input_ports[port_id].universe_address;
unsigned int buffer_size = sizeof(packet.data.dmx.data);
packet.data.dmx.length[0] = buffer_size >> 8;
packet.data.dmx.length[1] = buffer_size & 0xff;
buffer.Get(packet.data.dmx.data, &buffer_size);
// the packet size needs to be a multiple of two, correct here if needed
if (buffer_size % 2)
buffer_size++;
unsigned int size = sizeof(packet.data.dmx) - DMX_UNIVERSE_SIZE + buffer_size;
bool sent_ok = false;
if (m_input_ports[port_id].subscribed_nodes.size() >= BROADCAST_THRESHOLD) {
sent_ok = SendPacket(packet, size, m_interface.bcast_address);
m_input_ports[port_id].sequence_number++;
} else {
map<IPAddress, TimeStamp>::iterator iter =
m_input_ports[port_id].subscribed_nodes.begin();
TimeStamp last_heard_threshold = (
*m_plugin_adaptor->WakeUpTime() - TimeInterval(NODE_TIMEOUT, 0));
while (iter != m_input_ports[port_id].subscribed_nodes.end()) {
// if this node has timed out, remove it from the set
if (iter->second < last_heard_threshold) {
m_input_ports[port_id].subscribed_nodes.erase(iter++);
continue;
}
sent_ok |= SendPacket(packet, size, iter->first);
++iter;
}
if (!m_input_ports[port_id].subscribed_nodes.size()) {
OLA_DEBUG <<
"Suppressing data transmit due to no active nodes for universe" <<
static_cast<int>(m_input_ports[port_id].universe_address);
sent_ok = true;
} else {
// We sent at least one packet, increment the sequence number
m_input_ports[port_id].sequence_number++;
}
}
if (!sent_ok)
OLA_WARN << "Failed to send ArtNet DMX packet";
return sent_ok;
}
/*
* Send a TODRequest
* @param port_id port to send on
*/
bool ArtNetNode::SendTodRequest(uint8_t port_id) {
if (!CheckInputPortState(port_id, "ArtTodRequest"))
return false;
if (!GrabDiscoveryLock(port_id))
return true;
OLA_DEBUG << "Sending ArtTodRequest for address " <<
static_cast<int>(m_input_ports[port_id].universe_address);
artnet_packet packet;
PopulatePacketHeader(&packet, ARTNET_TODREQUEST);
memset(&packet.data.tod_request, 0, sizeof(packet.data.tod_request));
packet.data.tod_request.version = HostToNetwork(ARTNET_VERSION);
packet.data.tod_request.address_count = 1; // only one universe address
packet.data.tod_request.addresses[0] =
m_input_ports[port_id].universe_address;
unsigned int size = sizeof(packet.data.tod_request);
return SendPacket(packet, size, m_interface.bcast_address);
}
/*
* Flush the TOD and force discovery
* @param port_id port to send on
*/
bool ArtNetNode::ForceDiscovery(uint8_t port_id) {
if (!CheckInputPortState(port_id, "ArtTodControl"))
return false;
if (!GrabDiscoveryLock(port_id))
return true;
OLA_DEBUG << "Sending ArtTodControl";
artnet_packet packet;
PopulatePacketHeader(&packet, ARTNET_TODCONTROL);
memset(&packet.data.tod_control, 0, sizeof(packet.data.tod_control));
packet.data.tod_control.version = HostToNetwork(ARTNET_VERSION);
packet.data.tod_control.command = TOD_FLUSH_COMMAND;
packet.data.tod_control.address = m_input_ports[port_id].universe_address;
unsigned int size = sizeof(packet.data.tod_control);
return SendPacket(packet, size, m_interface.bcast_address);
}
/*
* Send an RDMRequest on this port, this may defer the sending if there are
* other outstanding messages in the queue.
* @param port_id the if of the port to send the request on
* @param request the RDMRequest object
*/
bool ArtNetNode::SendRDMRequest(uint8_t port_id, const RDMRequest *request) {
if (!CheckInputPortState(port_id, "ArtRDM"))
return false;
if (m_input_ports[port_id].pending_rdm_requests.size() >=
RDM_REQUEST_QUEUE_LIMIT) {
OLA_WARN << "RDM Request queue limit reached. " <<
m_input_ports[port_id].pending_rdm_requests.size() <<
" requests in queue, limit is " << RDM_REQUEST_QUEUE_LIMIT;
return false;
}
m_input_ports[port_id].pending_rdm_requests.push(request);
MaybeSendRDMRequest(port_id);
return true;
}
/*
* Set the RDM handlers for an Input port
* @param port_id the id of the port to set the handlers for
* @param on_tod the callback to be invoked when a ArtTod message is recieved,
* this callback is passed two UIDSets, one for the UIDs added and one for
* the UIDs that were removed
* @param on_rdm_response the callback to be invoked when a RDMResponse message
* is received. The callback is responsible for deleting the RDMResponse
* object.
*/
bool ArtNetNode::SetInputPortRDMHandlers(
uint8_t port_id,
rdm_tod_callback *on_tod,
rdm_response_callback *on_rdm_response) {
if (port_id > ARTNET_MAX_PORTS) {
OLA_WARN << "Port index of out bounds: " << port_id << " > " <<
ARTNET_MAX_PORTS;
return false;
}
if (m_input_ports[port_id].on_tod)
delete m_input_ports[port_id].on_tod;
if (m_input_ports[port_id].on_rdm_response)
delete m_input_ports[port_id].on_rdm_response;
m_input_ports[port_id].on_tod = on_tod;
m_input_ports[port_id].on_rdm_response = on_rdm_response;
return true;
}
/*
* Set the closure to be called when we receive data for this universe.
* @param universe the universe to register the handler for
* @param handler the Closure to call when there is data for this universe.
* Ownership of the closure is transferred to the node.
*/
bool ArtNetNode::SetDMXHandler(uint8_t port_id,
DmxBuffer *buffer,
Closure<void> *on_data) {
if (port_id > ARTNET_MAX_PORTS) {
OLA_WARN << "Port index of out bounds: " << port_id << " > " <<
ARTNET_MAX_PORTS;
return false;
}
if (m_output_ports[port_id].on_data)
delete m_output_ports[port_id].on_data;
m_output_ports[port_id].buffer = buffer;
m_output_ports[port_id].on_data = on_data;
return true;
}
/*
* Send an set of UIDs in one of more ArtTod packets
*/
bool ArtNetNode::SendTod(uint8_t port_id, const UIDSet &uid_set) {
if (!CheckOutputPortState(port_id, "ArtTodData"))
return false;
OLA_DEBUG << "Sending ArtTodRequest";
artnet_packet packet;
PopulatePacketHeader(&packet, ARTNET_TODDATA);
memset(&packet.data.tod_data, 0, sizeof(packet.data.tod_data));
packet.data.tod_data.version = HostToNetwork(ARTNET_VERSION);
packet.data.tod_data.rdm_version = RDM_VERSION;
packet.data.tod_data.port = 1 + port_id;
packet.data.tod_data.address = m_output_ports[port_id].universe_address;
uint16_t uids = std::min(uid_set.Size(),
(unsigned int) MAX_UIDS_PER_UNIVERSE);
packet.data.tod_data.uid_total = HostToNetwork(uids);
packet.data.tod_data.uid_count = ARTNET_MAX_UID_COUNT;
uint8_t (*ptr)[ola::rdm::UID::UID_SIZE] = packet.data.tod_data.tod;
unsigned int i = 0;
UIDSet::Iterator iter = uid_set.Begin();
while (iter != uid_set.End()) {
iter->Pack(*ptr, ola::rdm::UID::UID_SIZE);
i++;
iter++;
if (i % ARTNET_MAX_UID_COUNT == 0) {
packet.data.tod_data.block_count = i / ARTNET_MAX_UID_COUNT - 1;
SendPacket(packet,
sizeof(packet.data.tod_data),
m_interface.bcast_address);
ptr = packet.data.tod_data.tod;
} else {
ptr++;
}
}
if (i == 0 || i % ARTNET_MAX_UID_COUNT) {
packet.data.tod_data.uid_count = i % ARTNET_MAX_UID_COUNT;
packet.data.tod_data.block_count = i / ARTNET_MAX_UID_COUNT;
unsigned int size = sizeof(packet.data.tod_data) -
sizeof(packet.data.tod_data.tod) + i * ola::rdm::UID::UID_SIZE;
SendPacket(packet, size, m_interface.bcast_address);
}
return true;
}
/*
* Send a RDMResponse
* @param port_id the port id to send the response on
* @param response the RDMResponse to send
*/
bool ArtNetNode::SendRDMResponse(uint8_t port_id, const RDMResponse &response) {
if (!CheckOutputPortState(port_id, "ArtRDM"))
return false;
const UID &uid_destination = response.DestinationUID();
map<UID, IPAddress>::const_iterator iter =
m_output_ports[port_id].uid_map.find(uid_destination);
IPAddress destination = m_interface.bcast_address;
if (iter == m_output_ports[port_id].uid_map.end()) {
OLA_WARN << "Couldn't find " << uid_destination <<
" in the uid map, broadcasting packet";
} else {
destination = iter->second;
}
return SendRDMCommand(response,
destination,
m_output_ports[port_id].universe_address);
}
/*
* Set the RDM handlers for an Output port
*/
bool ArtNetNode::SetOutputPortRDMHandlers(
uint8_t port_id,
ola::Closure<void> *on_discover,
ola::Closure<void> *on_flush,
rdm_request_callback *on_rdm_request) {
if (port_id > ARTNET_MAX_PORTS) {
OLA_WARN << "Port index of out bounds: " << port_id << " > " <<
ARTNET_MAX_PORTS;
return false;
}
if (m_output_ports[port_id].on_discover)
delete m_output_ports[port_id].on_discover;
if (m_output_ports[port_id].on_flush)
delete m_output_ports[port_id].on_flush;
if (m_output_ports[port_id].on_rdm_request)
delete m_output_ports[port_id].on_rdm_request;
m_output_ports[port_id].on_discover = on_discover;
m_output_ports[port_id].on_flush = on_flush;
m_output_ports[port_id].on_rdm_request = on_rdm_request;
return true;
}
/*
* Called when there is data on this socket
*/
void ArtNetNode::SocketReady() {
artnet_packet packet;
ssize_t packet_size = sizeof(packet);
struct sockaddr_in source;
socklen_t source_length = sizeof(source);
if (!m_socket->RecvFrom(reinterpret_cast<uint8_t*>(&packet),
&packet_size,
source,
source_length))
return;
// skip packets sent by us
if (source.sin_addr.s_addr != m_interface.ip_address.s_addr)
HandlePacket(source.sin_addr, packet, packet_size);
}
/*
* Send an ArtPollReply message
*/
bool ArtNetNode::SendPollReply(const IPAddress &destination) {
artnet_packet packet;
PopulatePacketHeader(&packet, ARTNET_REPLY);
memset(&packet.data.reply, 0, sizeof(packet.data.reply));
memcpy(packet.data.reply.ip,
&m_interface.ip_address.s_addr,
sizeof(packet.data.reply.ip));
packet.data.reply.port = HostToLittleEndian(ARTNET_PORT);
packet.data.reply.subnet_address[1] = m_input_ports[0].universe_address >> 4;
packet.data.reply.oem = HostToNetwork(OEM_CODE);
packet.data.reply.status1 = 0xd2; // normal indicators, rdm enabled
packet.data.reply.esta_id = HostToLittleEndian(OPEN_LIGHTING_ESTA_CODE);
strncpy(reinterpret_cast<char*>(packet.data.reply.short_name),
m_short_name.data(),
ARTNET_SHORT_NAME_LENGTH);
strncpy(reinterpret_cast<char*>(packet.data.reply.long_name),
m_long_name.data(),
ARTNET_LONG_NAME_LENGTH);
std::stringstream str;
str << "#0001 [" << m_unsolicited_replies << "] OLA";
strncpy(reinterpret_cast<char*>(packet.data.reply.node_report),
str.str().data(),
ARTNET_REPORT_LENGTH);
packet.data.reply.number_ports[1] = ARTNET_MAX_PORTS;
for (unsigned int i = 0; i < ARTNET_MAX_PORTS; i++) {
packet.data.reply.port_types[i] = 0xc0; // input and output DMX
packet.data.reply.good_input[i] = m_input_ports[i].enabled ? 0x0 : 0x8;
packet.data.reply.sw_in[i] = m_input_ports[i].universe_address;
packet.data.reply.good_output[i] = (
(m_output_ports[i].enabled ? 0x80 : 0x00) |
(m_output_ports[i].merge_mode == ARTNET_MERGE_LTP ? 0x2 : 0x0) |
(m_output_ports[i].is_merging ? 0x8 : 0x0));
packet.data.reply.sw_out[i] = m_output_ports[i].universe_address;
}
packet.data.reply.style = NODE_CODE;
memcpy(packet.data.reply.mac,
m_interface.hw_address,
ola::network::MAC_LENGTH);
memcpy(packet.data.reply.bind_ip,
&m_interface.ip_address.s_addr,
sizeof(packet.data.reply.bind_ip));
// maybe set status2 here if the web UI is enabled
if (!SendPacket(packet, sizeof(packet.data.reply), destination)) {
OLA_INFO << "Failed to send ArtPollReply";
return false;
}
return true;
}
/*
* Send an IPProgReply
*/
bool ArtNetNode::SendIPReply(const IPAddress &destination) {
artnet_packet packet;
PopulatePacketHeader(&packet, ARTNET_REPLY);
memset(&packet.data.ip_reply, 0, sizeof(packet.data.ip_reply));
packet.data.ip_reply.version = HostToNetwork(ARTNET_VERSION);
memcpy(packet.data.ip_reply.ip,
&m_interface.ip_address.s_addr,
sizeof(packet.data.ip_reply.ip));
memcpy(packet.data.ip_reply.subnet,
&m_interface.ip_address.s_addr,
sizeof(packet.data.ip_reply.subnet));
packet.data.ip_reply.port = HostToLittleEndian(ARTNET_PORT);
if (!SendPacket(packet, sizeof(packet.data.ip_reply), destination)) {
OLA_INFO << "Failed to send ArtIpProgReply";
return false;
}
return true;
}
/*
* Handle a artnet packet
*/
void ArtNetNode::HandlePacket(const IPAddress &source_address,
const artnet_packet &packet,
unsigned int packet_size) {
unsigned int header_size = sizeof(packet) - sizeof(packet.data);
if (packet_size <= header_size) {
OLA_WARN << "Skipping small artnet packet received, size=" << packet_size;
return;
}
switch (LittleEndianToHost(packet.op_code)) {
case ARTNET_POLL:
HandlePollPacket(source_address,
packet.data.poll,
packet_size - header_size);
break;
case ARTNET_REPLY:
HandleReplyPacket(source_address,
packet.data.reply,
packet_size - header_size);
break;
case ARTNET_DMX:
HandleDataPacket(source_address,
packet.data.dmx,
packet_size - header_size);
break;
case ARTNET_TODREQUEST:
HandleTodRequest(source_address,
packet.data.tod_request,
packet_size - header_size);
break;
case ARTNET_TODDATA:
HandleTodData(source_address,
packet.data.tod_data,
packet_size - header_size);
break;
case ARTNET_TODCONTROL:
HandleTodControl(source_address,
packet.data.tod_control,
packet_size - header_size);
break;
case ARTNET_RDM:
HandleRdm(source_address,
packet.data.rdm,
packet_size - header_size);
break;
case ARTNET_IP_PROGRAM:
HandleIPProgram(source_address,
packet.data.ip_program,
packet_size - header_size);
break;
case ARTNET_RDM_SUB:
// Not implemented
break;
default:
OLA_INFO << "ArtNet got unknown packet " << std::hex <<
LittleEndianToHost(packet.op_code);
}
}
/*
* Handle an ArtPoll packet
*/
void ArtNetNode::HandlePollPacket(const IPAddress &source_address,
const artnet_poll_t &packet,
unsigned int packet_size) {
if (!CheckPacketSize(source_address, "ArtPoll", packet_size, sizeof(packet)))
return;
if (!CheckPacketVersion(source_address, "ArtPoll", packet.version))
return;
m_send_reply_on_change = packet.talk_to_me & 0x02;
// It's unclear if this should be broadcast or unicast, stick with broadcast
SendPollReply(m_interface.bcast_address);
(void) source_address;
}
/*
* Handle an ArtPollReply packet
*/
void ArtNetNode::HandleReplyPacket(const IPAddress &source_address,
const artnet_reply_t &packet,
unsigned int packet_size) {
// older versions don't have the bind_ip and the extra filler, make sure we
// support these
unsigned int minimum_reply_size = (
sizeof(packet) -
sizeof(packet.filler) -
sizeof(packet.status2) -
sizeof(packet.bind_index) -
sizeof(packet.bind_ip));
if (!CheckPacketSize(source_address, "ArtPollReply", packet_size,
minimum_reply_size))
return;
// Update the subscribed nodes list
unsigned int port_limit = std::min((uint8_t) ARTNET_MAX_PORTS,
packet.number_ports[1]);
for (unsigned int i = 0; i < port_limit; i++) {
if (packet.port_types[i] & 0x80) {
// port is of type output
uint8_t universe_id = packet.sw_out[i];
for (unsigned int port_id = 0; port_id < ARTNET_MAX_PORTS; port_id++) {
if (m_input_ports[port_id].enabled &&
m_input_ports[port_id].universe_address == universe_id) {
m_input_ports[port_id].subscribed_nodes[source_address] =
*m_plugin_adaptor->WakeUpTime();
}
}
}
}
}
/*
* Handle a DMX Data packet, this takes care of the merging
*/
void ArtNetNode::HandleDataPacket(const IPAddress &source_address,
const artnet_dmx_t &packet,
unsigned int packet_size) {
// The data section needs to be at least 2 bytes according to the spec
unsigned int header_size = sizeof(artnet_dmx_t) - DMX_UNIVERSE_SIZE;
if (!CheckPacketSize(source_address, "ArtDmx", packet_size, header_size + 2))
return;
if (!CheckPacketVersion(source_address, "ArtDmx", packet.version))
return;
uint16_t universe_id = LittleEndianToHost(packet.universe);
uint16_t data_size = std::min(
(unsigned int) ((packet.length[0] << 8) + packet.length[1]),
packet_size - header_size);
for (unsigned int port_id = 0; port_id < ARTNET_MAX_PORTS; port_id++) {
if (m_output_ports[port_id].enabled &&
m_output_ports[port_id].universe_address == universe_id &&
m_output_ports[port_id].on_data &&
m_output_ports[port_id].buffer) {
// update this port, doing a merge if necessary
DMXSource source;
source.address = source_address;
source.timestamp = *m_plugin_adaptor->WakeUpTime();
source.buffer.Set(packet.data, data_size);
UpdatePortFromSource(&m_output_ports[port_id], source);
}
}
}
/*
* Handle a TOD Request packet
*/
void ArtNetNode::HandleTodRequest(const IPAddress &source_address,
const artnet_todrequest_t &packet,
unsigned int packet_size) {
unsigned int header_size = sizeof(packet) - sizeof(packet.addresses);
if (!CheckPacketSize(source_address, "ArtTodRequest", packet_size,
header_size))
return;
if (!CheckPacketVersion(source_address, "ArtTodRequest", packet.version))
return;
if (packet.command) {
OLA_INFO << "ArtTodRequest received but command field was " <<
static_cast<int>(packet.command);
return;
}
unsigned int addresses = std::min(
packet_size - header_size,
static_cast<unsigned int>(packet.address_count));
addresses = std::min(
static_cast<unsigned int>(ARTNET_MAX_RDM_ADDRESS_COUNT),
addresses);
bool handler_called[ARTNET_MAX_PORTS];
memset(handler_called, 0, sizeof(handler_called));
for (unsigned int i = 0; i < addresses; i++) {
for (unsigned int port_id = 0; port_id < ARTNET_MAX_PORTS; port_id++) {
if (m_output_ports[port_id].enabled &&
m_output_ports[port_id].universe_address == packet.addresses[i] &&
m_output_ports[port_id].on_discover &&
!handler_called[port_id]) {
m_output_ports[port_id].on_discover->Run();
handler_called[port_id] = true;
}
}
}
}
/*
* Handle a TOD data packet
*/
void ArtNetNode::HandleTodData(const IPAddress &source_address,
const artnet_toddata_t &packet,
unsigned int packet_size) {
unsigned int expected_size = sizeof(packet) - sizeof(packet.tod);
if (!CheckPacketSize(source_address, "ArtTodData", packet_size,
expected_size))
return;
if (!CheckPacketVersion(source_address, "ArtTodData", packet.version))
return;
if (packet.rdm_version != RDM_VERSION) {
OLA_WARN << "Dropping non standard RDM version: " <<
static_cast<int>(packet.rdm_version);
return;
}
if (packet.command_response) {
OLA_WARN << "Command response 0x" << std::hex << packet.command_response <<
" != 0x0";
return;
}
for (unsigned int port_id = 0; port_id < ARTNET_MAX_PORTS; port_id++) {
if (m_input_ports[port_id].enabled &&
m_input_ports[port_id].universe_address == packet.address) {
UpdatePortFromTodPacket(port_id, source_address, packet, packet_size);
}
}
}
/*
* Handle a TOD Control packet
*/
void ArtNetNode::HandleTodControl(const IPAddress &source_address,
const artnet_todcontrol_t &packet,
unsigned int packet_size) {
if (!CheckPacketSize(source_address, "ArtTodControl", packet_size,
sizeof(packet)))
return;
if (!CheckPacketVersion(source_address, "ArtTodControl", packet.version))
return;
if (packet.command != TOD_FLUSH_COMMAND)
return;
for (unsigned int port_id = 0; port_id < ARTNET_MAX_PORTS; port_id++) {
if (m_output_ports[port_id].enabled &&
m_output_ports[port_id].universe_address == packet.address &&
m_output_ports[port_id].on_flush) {
m_output_ports[port_id].on_flush->Run();
}
}
}
/*
* Handle an RDM packet
*/
void ArtNetNode::HandleRdm(const IPAddress &source_address,
const artnet_rdm_t &packet,
unsigned int packet_size) {
unsigned int header_size = sizeof(packet) - ARTNET_MAX_RDM_DATA;
if (!CheckPacketSize(source_address, "ArtRDM", packet_size, header_size))
return;
if (!CheckPacketVersion(source_address, "ArtRDM", packet.version))
return;
if (packet.rdm_version != RDM_VERSION) {
OLA_INFO << "Dropping non standard RDM version: " <<
static_cast<int>(packet.rdm_version);
return;
}
if (packet.command) {
OLA_WARN << "Unknown RDM command " << static_cast<int>(packet.command);
return;
}
unsigned int rdm_length = packet_size - header_size;
if (!rdm_length)
return;
// look for the port that this was sent to, once we know the port we can try
// to parse the message
for (unsigned int port_id = 0; port_id < ARTNET_MAX_PORTS; port_id++) {
if (m_output_ports[port_id].enabled &&
m_output_ports[port_id].universe_address == packet.address &&
m_output_ports[port_id].on_rdm_request) {
RDMRequest *request = RDMRequest::InflateFromData(packet.data,
rdm_length);
if (request) {
// update the output port uid map
m_output_ports[port_id].uid_map[request->SourceUID()] = source_address;
m_output_ports[port_id].on_rdm_request->Run(request);
}
}
if (m_input_ports[port_id].enabled &&
m_input_ports[port_id].universe_address == packet.address &&
m_input_ports[port_id].on_rdm_response) {
RDMResponse *response = RDMResponse::InflateFromData(packet.data,
rdm_length);
if (response)
HandleRdmResponse(port_id, response);
}
}
}
/**
* Handle an RDM response, taking care to deal with ACK_OVERFLOW messages.
* <rant>
* ArtNet as a protocol is broken, the nodes don't buffer ACK_OVERFLOW messages
* so if another GET/SET message arrives from *any* controller the ACK_OVERFLOW
* session will be reset, causing the controller to spin in a loop.
* TODO(simonn): lock out sending when we see an ACK_OVERFLOW to give us a
* better chance of completing the session.
* </rant>
*/
void ArtNetNode::HandleRdmResponse(unsigned int port_id,
const RDMResponse *response) {
InputPort &input_port = m_input_ports[port_id];
if (input_port.pending_rdm_requests.empty())
return;
const RDMRequest *request = input_port.pending_rdm_requests.front();
if (request->SourceUID() != response->DestinationUID() ||
request->DestinationUID() != response->SourceUID() ||
request->SubDevice() != response->SubDevice() ||
request->ParamId() != response->ParamId()) {
OLA_INFO << "Got an unexpected RDM response";
return;
}
if ((request->CommandClass() == RDMCommand::GET_COMMAND &&
response->CommandClass() != RDMCommand::GET_COMMAND_RESPONSE) ||
(request->CommandClass() == RDMCommand::SET_COMMAND &&
response->CommandClass() != RDMCommand::SET_COMMAND_RESPONSE)) {
OLA_INFO << "Unmatched RDM response, request CC was " << std::hex <<
static_cast<int>(request->CommandClass()) << ", response CC was " <<
static_cast<int>(response->CommandClass());
return;
}
if (input_port.overflowed_response) {
if (response->ResponseType() == ola::rdm::ACK_OVERFLOW) {
OLA_INFO << "Continued ACK_OVERFLOW session for " <<
response->SourceUID() << " -> " << response->DestinationUID() <<
", PID " << std::hex << response->ParamId();
const RDMResponse *existing_reponse = input_port.overflowed_response;
input_port.overflowed_response = RDMResponse::CombineResponses(
existing_reponse,
response);
delete response;
delete existing_reponse;
if (input_port.overflowed_response) {
// combine worked
if (SendFirstRDMRequest(port_id)) {
input_port.overflowed_response = response;
return;
} else {
// send failed, abort
delete input_port.overflowed_response;
input_port.overflowed_response = NULL;
}
}
} else if (response->ResponseType() == ola::rdm::ACK) {
// end of the session, remove and call handler
OLA_INFO << "Final packet in ACK_OVERFLOW session for " <<
response->SourceUID() << " -> " << response->DestinationUID() <<
", PID " << std::hex << response->ParamId();
const RDMResponse *full_response = RDMResponse::CombineResponses(
input_port.overflowed_response,
response);
delete input_port.overflowed_response;
input_port.overflowed_response = NULL;
delete response;
if (full_response)
input_port.on_rdm_response->Run(full_response);
} else {
// not an ack or overflow, abort transfer
OLA_INFO << "Got " << static_cast<int>(response->ResponseType()) <<
" in ACK_OVERFLOW session for " <<
response->SourceUID() << " -> " << response->DestinationUID() <<
", PID " << std::hex << response->ParamId() << ", aborting session";
delete response;
}
} else if (response->ResponseType() == ola::rdm::ACK_OVERFLOW) {
OLA_INFO << "Got new ACK_OVERFLOW for " <<
response->SourceUID() << " -> " << response->DestinationUID() <<
", PID " << std::hex << response->ParamId();
if (SendFirstRDMRequest(port_id)) {
input_port.overflowed_response = response;
return;
} else {
// send failed, abort
delete response;
}
} else {
// this is just a normal response
input_port.on_rdm_response->Run(response);
}
ClearPendingRDMRequest(port_id);
MaybeSendRDMRequest(port_id);
}
/**
* Handle an IP Program message.
*/
void ArtNetNode::HandleIPProgram(const IPAddress &source_address,
const artnet_ip_prog_t &packet,
unsigned int packet_size) {
if (!CheckPacketSize(source_address, "ArtIpProg", packet_size,
sizeof(packet)))
return;
if (!CheckPacketVersion(source_address, "ArtIpProg", packet.version))
return;
OLA_INFO <<
"Got ArtIpProgram, ignoring because we don't support remote configuration";
}
/*
* Fill in the header for a packet
*/
void ArtNetNode::PopulatePacketHeader(artnet_packet *packet,
uint16_t op_code) {
strncpy(reinterpret_cast<char*>(packet->id), ARTNET_ID, sizeof(packet->id));
packet->op_code = HostToLittleEndian(op_code);
}
/*
* Increment the counts for all the uids
*/
void ArtNetNode::IncrementUIDCounts(uint8_t port_id) {
uid_map::iterator iter = m_input_ports[port_id].uids.begin();
for (; iter != m_input_ports[port_id].uids.end(); ++iter) {
iter->second.second++;
}
}
/*
* Send an ArtNet packet
* @param packet
* @param size the size of the packet, excluding the header portion
* @param destination where to send the packet to
*/
bool ArtNetNode::SendPacket(const artnet_packet &packet,
unsigned int size,
const IPAddress &ip_destination) {
struct sockaddr_in destination;
destination.sin_family = AF_INET;
destination.sin_port = HostToNetwork(ARTNET_PORT);
destination.sin_addr = ip_destination;
size += sizeof(packet.id) + sizeof(packet.op_code);
unsigned int bytes_sent = m_socket->SendTo(
reinterpret_cast<const uint8_t*>(&packet),
size,
destination);
if (bytes_sent != size) {
OLA_WARN << "Only sent " << bytes_sent << " of " << size;
return false;
}
return true;
}
/**
* Send an pending RDM request if we can.
* @param port_id the id of the port to send the requests for.
*/
void ArtNetNode::MaybeSendRDMRequest(uint8_t port_id) {
InputPort &input_port = m_input_ports[port_id];
if (input_port.pending_rdm_requests.empty())
return;
if (input_port.rdm_send_timeout != ola::network::INVALID_TIMEOUT)
return;
if (SendFirstRDMRequest(port_id)) {
input_port.rdm_send_timeout = m_plugin_adaptor->RegisterSingleTimeout(
RDM_REQUEST_TIMEOUT_MS,
ola::NewSingleClosure(this, &ArtNetNode::TimeoutRDMRequest, port_id));
} else {
// send failed, remove this request and try the next one.
ClearPendingRDMRequest(port_id);
MaybeSendRDMRequest(port_id);
}
}
/**
* Send the RDM request from the head of the queue.
* @param port_id the id of the port to send the requests for.
*/
bool ArtNetNode::SendFirstRDMRequest(uint8_t port_id) {
InputPort &input_port = m_input_ports[port_id];
const RDMRequest *request = input_port.pending_rdm_requests.front();
const UID &uid_destination = request->DestinationUID();
uid_map::const_iterator iter =
m_input_ports[port_id].uids.find(uid_destination);
IPAddress destination = m_interface.bcast_address;
if (iter == m_input_ports[port_id].uids.end()) {
OLA_WARN << "Couldn't find " << uid_destination <<
" in the uid map, broadcasting packet";
} else {
destination = iter->second.first;
}
return SendRDMCommand(*request,
destination,
m_input_ports[port_id].universe_address);
}
/**
* Timeout a pending RDM request
* @param port_id the id of the port to timeout.
*/
void ArtNetNode::TimeoutRDMRequest(uint8_t port_id) {
OLA_INFO << "RDM Request timed out.";
m_input_ports[port_id].rdm_send_timeout = ola::network::INVALID_TIMEOUT;
ClearPendingRDMRequest(port_id);
MaybeSendRDMRequest(port_id);
}
/**
* Remove the head of the pending queue and reset the timeout.
*/
void ArtNetNode::ClearPendingRDMRequest(uint8_t port_id) {
InputPort &input_port = m_input_ports[port_id];
if (!input_port.pending_rdm_requests.empty()) {
delete input_port.pending_rdm_requests.front();
input_port.pending_rdm_requests.pop();
}
if (input_port.rdm_send_timeout != ola::network::INVALID_TIMEOUT) {
m_plugin_adaptor->RemoveTimeout(input_port.rdm_send_timeout);
input_port.rdm_send_timeout = ola::network::INVALID_TIMEOUT;
}
}
/*
* Send a generic ArtRdm message
*/
bool ArtNetNode::SendRDMCommand(const RDMCommand &command,
const IPAddress &destination,
uint8_t universe) {
artnet_packet packet;
PopulatePacketHeader(&packet, ARTNET_RDM);
memset(&packet.data.rdm, 0, sizeof(packet.data.rdm));
packet.data.rdm.version = HostToNetwork(ARTNET_VERSION);
packet.data.rdm.rdm_version = RDM_VERSION;
packet.data.rdm.address = universe;
unsigned int rdm_size = ARTNET_MAX_RDM_DATA;
command.Pack(packet.data.rdm.data, &rdm_size);
unsigned int packet_size = sizeof(packet.data.rdm) - ARTNET_MAX_RDM_DATA +
rdm_size;
return SendPacket(packet, packet_size, destination);
}
/*
* Update a port from a source, merging if necessary
*/
void ArtNetNode::UpdatePortFromSource(OutputPort *port,
const DMXSource &source) {
TimeStamp merge_time_threshold = (
*m_plugin_adaptor->WakeUpTime() - TimeInterval(MERGE_TIMEOUT, 0));
unsigned int first_empty_slot = MAX_MERGE_SOURCES;
unsigned int source_slot = MAX_MERGE_SOURCES;
unsigned int active_sources = 0;
for (unsigned int i = 0; i < MAX_MERGE_SOURCES; i++) {
if (port->sources[i].address.s_addr == source.address.s_addr)
source_slot = i;
// timeout old sources
if (port->sources[i].timestamp < merge_time_threshold)
port->sources[i].address.s_addr = 0;
if (port->sources[i].address.s_addr != 0)
active_sources++;
else if (i < first_empty_slot)
first_empty_slot = i;
}
if (source_slot == MAX_MERGE_SOURCES) {
if (first_empty_slot == MAX_MERGE_SOURCES) {
// No room at the inn
OLA_WARN << "Max merge sources reached, ignoring";
return;
}
if (active_sources == 0) {
port->is_merging = false;
} else {
OLA_INFO << "Entered merge mode for universe " <<
static_cast<int>(port->universe_address);
port->is_merging = true;
if (m_send_reply_on_change) {
m_unsolicited_replies++;
SendPollReply(m_interface.bcast_address);
}
}
source_slot = first_empty_slot;
} else if (active_sources == 1) {
port->is_merging = false;
}
port->sources[source_slot] = source;
// Now we need to merge
if (port->merge_mode == ARTNET_MERGE_LTP) {
// the current source is the latest
(*port->buffer) = source.buffer;
} else {
// HTP merge
bool first = true;
for (unsigned int i = 0; i < MAX_MERGE_SOURCES; i++) {
if (port->sources[i].address.s_addr != 0) {
if (first) {
(*port->buffer) = port->sources[i].buffer;
first = false;
} else {
port->buffer->HTPMerge(port->sources[i].buffer);
}
}
}
}
port->on_data->Run();
}
/*
* Check the version number of a incomming packet
*/
bool ArtNetNode::CheckPacketVersion(const IPAddress &source_address,
const string &packet_type,
uint16_t version) {
if (NetworkToHost(version) != ARTNET_VERSION) {
OLA_INFO << packet_type << " version mismatch, was " <<
NetworkToHost(version) << " from " <<
AddressToString(source_address);
return false;
}
return true;
}
/*
* Check the size of an incoming packet
*/
bool ArtNetNode::CheckPacketSize(const IPAddress &source_address,
const string &packet_type,
unsigned int actual_size,
unsigned int expected_size) {
if (actual_size < expected_size) {
OLA_INFO << packet_type << " from " << AddressToString(source_address) <<
" was too small, got " << actual_size <<
" required at least " << expected_size;
return false;
}
return true;
}
/*
* Check if the input port is available for sending
*/
bool ArtNetNode::CheckInputPortState(uint8_t port_id, const string &action) {
return CheckPortState(port_id, action, false);
}
/*
* Check if the output port is available for sending
*/
bool ArtNetNode::CheckOutputPortState(uint8_t port_id, const string &action) {
return CheckPortState(port_id, action, true);
}
/*
* Check if a port is available for sending
*/
bool ArtNetNode::CheckPortState(uint8_t port_id,
const string &action,
bool is_output) {
if (!m_running)
return false;
if (port_id > ARTNET_MAX_PORTS) {
OLA_WARN << "Port index of out bounds: " << port_id << " > " <<
ARTNET_MAX_PORTS;
return false;
}
if ((is_output && !m_output_ports[port_id].enabled) ||
(!is_output && !m_input_ports[port_id].enabled)) {
OLA_INFO << "Attempt to send " << action << " on an inactive port";
return false;
}
return true;
}
/*
* Setup the networking components.
*/
bool ArtNetNode::InitNetwork() {
m_socket = new UdpSocket();
if (!m_socket->Init()) {
OLA_WARN << "Socket init failed";
delete m_socket;
return false;
}
if (!m_socket->Bind(ARTNET_PORT)) {
OLA_WARN << "Failed to bind to:" << ARTNET_PORT;
delete m_socket;
return false;
}
if (!m_socket->EnableBroadcast()) {
OLA_WARN << "Failed to enable broadcasting";
delete m_socket;
return false;
}
m_socket->SetOnData(NewClosure(this, &ArtNetNode::SocketReady));
m_plugin_adaptor->AddSocket(m_socket);
return true;
}
/*
* Update a port with a new TOD list
*/
void ArtNetNode::UpdatePortFromTodPacket(uint8_t port_id,
const IPAddress &source_address,
const artnet_toddata_t &packet,
unsigned int packet_size) {
unsigned int tod_size = packet_size - (sizeof(packet) - sizeof(packet.tod));
unsigned int uid_count = std::min(tod_size / ola::rdm::UID::UID_SIZE,
(unsigned int) packet.uid_count);
OLA_DEBUG << "Got TOD data packet with " << uid_count << " uids";
bool changed = false;
uid_map &port_uids = m_input_ports[port_id].uids;
UIDSet uid_set;
for (unsigned int i = 0; i < uid_count; i++) {
UID uid(packet.tod[i]);
uid_set.AddUID(uid);
uid_map::iterator iter = port_uids.find(uid);
if (iter == port_uids.end()) {
port_uids[uid] = std::pair<IPAddress, uint8_t>(source_address, 0);
changed = true;
} else {
if (iter->second.first.s_addr != source_address.s_addr) {
OLA_WARN << "UID " << uid << " changed from " <<
AddressToString(iter->second.first) << " to " <<
AddressToString(source_address);
iter->second.first = source_address;
}
iter->second.second = 0;
}
}
// If this is the one and only block from this node, we can remove all uids
// that don't appear in it.
// There is a bug in ArtNet nodes where sometimes UidCount > UidTotal.
if (uid_count >= NetworkToHost(packet.uid_total)) {
uid_map::iterator iter = port_uids.begin();
while (iter != port_uids.end()) {
if (iter->second.first.s_addr == source_address.s_addr &&
!uid_set.Contains(iter->first)) {
port_uids.erase(iter++);
changed = true;
} else {
++iter;
}
}
}
// Removing uids from multi-block messages is much harder as you need to
// consider dropped packets. For the moment we rely on the
// RDM_MISSED_TODDATA_LIMIT to clean these up.
// TODO(simon): figure this out sometime
if (changed)
NotifyClientOfNewTod(port_id);
}
/*
* Start the discovery process, this puts the port into discovery mode and
* sets up the completion callback
*/
bool ArtNetNode::GrabDiscoveryLock(uint8_t port_id) {
if (m_input_ports[port_id].discovery_running) {
OLA_INFO <<
"ArtNet UID discovery already running, ignoring additional requests";
return false;
}
m_input_ports[port_id].discovery_running = true;
// increment the count of all current uids
uid_map::iterator iter = m_input_ports[port_id].uids.begin();
for (; iter != m_input_ports[port_id].uids.end(); ++iter) {
iter->second.second++;
}
m_discovery_timeout = m_plugin_adaptor->RegisterSingleTimeout(
RDM_TOD_TIMEOUT_MS,
ola::NewSingleClosure(this, &ArtNetNode::ReleaseDiscoveryLock, port_id));
return true;
}
/*
* Called when the discovery process times out.
*/
void ArtNetNode::ReleaseDiscoveryLock(uint8_t port_id) {
OLA_INFO << "Discovery process timeout";
m_discovery_timeout = ola::network::INVALID_TIMEOUT;
// delete all uids that have reached the max count
bool changed = false;
uid_map::iterator iter = m_input_ports[port_id].uids.begin();
while (iter != m_input_ports[port_id].uids.end()) {
if (iter->second.second == RDM_MISSED_TODDATA_LIMIT) {
changed = true;
m_input_ports[port_id].uids.erase(iter++);
} else {
++iter;
}
}
if (changed) {
OLA_INFO << "Some uids have timed out, updating.";
NotifyClientOfNewTod(port_id);
}
m_input_ports[port_id].discovery_running = false;
}
/*
* Notify the client of a new TOD
*/
void ArtNetNode::NotifyClientOfNewTod(uint8_t port_id) {
if (m_input_ports[port_id].on_tod) {
uid_map &port_uids = m_input_ports[port_id].uids;
UIDSet uids;
uid_map::iterator uid_iter = port_uids.begin();
for (; uid_iter != port_uids.end(); ++uid_iter)
uids.AddUID(uid_iter->first);
m_input_ports[port_id].on_tod->Run(uids);
}
}
} // artnet
} // plugin
} // ola