Merge pull request #6398 from ZappoMan/messaging

messages-mixer

assignment-client/src/Agent.cpp

@@ -17,6 +17,7 @@
 #include <QtNetwork/QNetworkReply>
 
 #include <AvatarHashMap.h>
+#include <MessagesClient.h>
 #include <NetworkAccessManager.h>
 #include <NodeList.h>
 #include <udt/PacketHeaders.h>
@@ -93,7 +94,7 @@ void Agent::handleJurisdictionPacket(QSharedPointer<NLPacket> packet, SharedNode
         DependencyManager::get<EntityScriptingInterface>()->getJurisdictionListener()->
             queueReceivedPacket(packet, senderNode);
     }
-} 
+}
 
 void Agent::handleAudioPacket(QSharedPointer<NLPacket> packet) {
     _receivedAudioStream.parseData(*packet);
@@ -109,11 +110,21 @@ const int PING_INTERVAL = 1000;
 void Agent::run() {
     ThreadedAssignment::commonInit(AGENT_LOGGING_NAME, NodeType::Agent);
 
+    // Setup MessagesClient
+    auto messagesClient = DependencyManager::set<MessagesClient>();
+    QThread* messagesThread = new QThread;
+    messagesThread->setObjectName("Messages Client Thread");
+    messagesClient->moveToThread(messagesThread);
+    connect(messagesThread, &QThread::started, messagesClient.data(), &MessagesClient::init);
+    messagesThread->start();
+
+
     auto nodeList = DependencyManager::get<NodeList>();
     nodeList->addSetOfNodeTypesToNodeInterestSet(NodeSet()
                                                  << NodeType::AudioMixer
                                                  << NodeType::AvatarMixer
                                                  << NodeType::EntityServer
+                                                 << NodeType::MessagesMixer
                                                 );
 
     _pingTimer = new QTimer(this);

assignment-client/src/AssignmentFactory.cpp

@@ -17,6 +17,7 @@
 #include "avatars/AvatarMixer.h"
 #include "entities/EntityServer.h"
 #include "assets/AssetServer.h"
+#include "messages/MessagesMixer.h"
 
 ThreadedAssignment* AssignmentFactory::unpackAssignment(NLPacket& packet) {
 
@@ -36,6 +37,8 @@ ThreadedAssignment* AssignmentFactory::unpackAssignment(NLPacket& packet) {
             return new EntityServer(packet);
         case Assignment::AssetServerType:
             return new AssetServer(packet);
+        case Assignment::MessagesMixerType:
+            return new MessagesMixer(packet);
         default:
             return NULL;
     }

assignment-client/src/entities/EntityServer.cpp

@@ -112,7 +112,6 @@ int EntityServer::sendSpecialPackets(const SharedNodePointer& node, OctreeQueryN
         quint64 deletePacketSentAt = usecTimestampNow();
         EntityTreePointer tree = std::static_pointer_cast<EntityTree>(_tree);
         auto recentlyDeleted = tree->getRecentlyDeletedEntityIDs();
-        bool hasMoreToSend = true;
 
         packetsSent = 0;
 

assignment-client/src/messages/MessagesMixer.cpp

@@ -0,0 +1,155 @@
+//
+//  MessagesMixer.cpp
+//  assignment-client/src/messages
+//
+//  Created by Brad hefta-Gaub on 11/16/2015.
+//  Copyright 2015 High Fidelity, Inc.
+//
+//  Distributed under the Apache License, Version 2.0.
+//  See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html
+//
+
+#include <QtCore/QCoreApplication>
+#include <QtCore/QJsonObject>
+#include <QBuffer>
+
+#include <LogHandler.h>
+#include <NodeList.h>
+#include <udt/PacketHeaders.h>
+
+#include "MessagesMixer.h"
+
+const QString MESSAGES_MIXER_LOGGING_NAME = "messages-mixer";
+
+MessagesMixer::MessagesMixer(NLPacket& packet) :
+    ThreadedAssignment(packet)
+{
+    // make sure we hear about node kills so we can tell the other nodes
+    connect(DependencyManager::get<NodeList>().data(), &NodeList::nodeKilled, this, &MessagesMixer::nodeKilled);
+
+    auto& packetReceiver = DependencyManager::get<NodeList>()->getPacketReceiver();
+    packetReceiver.registerMessageListener(PacketType::MessagesData, this, "handleMessages");
+    packetReceiver.registerMessageListener(PacketType::MessagesSubscribe, this, "handleMessagesSubscribe");
+    packetReceiver.registerMessageListener(PacketType::MessagesUnsubscribe, this, "handleMessagesUnsubscribe");
+}
+
+MessagesMixer::~MessagesMixer() {
+}
+
+void MessagesMixer::nodeKilled(SharedNodePointer killedNode) {
+    // FIXME - remove the node from the subscription maps
+}
+
+void MessagesMixer::handleMessages(QSharedPointer<NLPacketList> packetList, SharedNodePointer senderNode) {
+    Q_ASSERT(packetList->getType() == PacketType::MessagesData);
+
+    QByteArray packetData = packetList->getMessage();
+    QBuffer packet{ &packetData };
+    packet.open(QIODevice::ReadOnly);
+
+    quint16 channelLength;
+    packet.read(reinterpret_cast<char*>(&channelLength), sizeof(channelLength));
+    auto channelData = packet.read(channelLength);
+    QString channel = QString::fromUtf8(channelData);
+
+    quint16 messageLength;
+    packet.read(reinterpret_cast<char*>(&messageLength), sizeof(messageLength));
+    auto messageData = packet.read(messageLength);
+    QString message = QString::fromUtf8(messageData);
+    
+    auto nodeList = DependencyManager::get<NodeList>();
+
+    nodeList->eachMatchingNode(
+        [&](const SharedNodePointer& node)->bool {
+
+        return node->getType() == NodeType::Agent && node->getActiveSocket() &&
+                _channelSubscribers[channel].contains(node->getUUID());
+    },
+        [&](const SharedNodePointer& node) {
+
+        auto packetList = NLPacketList::create(PacketType::MessagesData, QByteArray(), true, true);
+
+        auto channelUtf8 = channel.toUtf8();
+        quint16 channelLength = channelUtf8.length();
+        packetList->writePrimitive(channelLength);
+        packetList->write(channelUtf8);
+
+        auto messageUtf8 = message.toUtf8();
+        quint16 messageLength = messageUtf8.length();
+        packetList->writePrimitive(messageLength);
+        packetList->write(messageUtf8);
+
+        nodeList->sendPacketList(std::move(packetList), *node);
+    });
+}
+
+void MessagesMixer::handleMessagesSubscribe(QSharedPointer<NLPacketList> packetList, SharedNodePointer senderNode) {
+    Q_ASSERT(packetList->getType() == PacketType::MessagesSubscribe);
+    QString channel = QString::fromUtf8(packetList->getMessage());
+    qDebug() << "Node [" << senderNode->getUUID() << "] subscribed to channel:" << channel;
+    _channelSubscribers[channel] << senderNode->getUUID();
+}
+
+void MessagesMixer::handleMessagesUnsubscribe(QSharedPointer<NLPacketList> packetList, SharedNodePointer senderNode) {
+    Q_ASSERT(packetList->getType() == PacketType::MessagesUnsubscribe);
+    QString channel = QString::fromUtf8(packetList->getMessage());
+    qDebug() << "Node [" << senderNode->getUUID() << "] unsubscribed from channel:" << channel;
+
+    if (_channelSubscribers.contains(channel)) {
+        _channelSubscribers[channel].remove(senderNode->getUUID());
+    }
+}
+
+// FIXME - make these stats relevant
+void MessagesMixer::sendStatsPacket() {
+    QJsonObject statsObject;
+    QJsonObject messagesObject;
+    auto nodeList = DependencyManager::get<NodeList>();
+    // add stats for each listerner
+    nodeList->eachNode([&](const SharedNodePointer& node) {
+        QJsonObject messagesStats;
+
+        // add the key to ask the domain-server for a username replacement, if it has it
+        messagesStats[USERNAME_UUID_REPLACEMENT_STATS_KEY] = uuidStringWithoutCurlyBraces(node->getUUID());
+        messagesStats["outbound_kbps"] = node->getOutboundBandwidth();
+        messagesStats["inbound_kbps"] = node->getInboundBandwidth();
+
+        messagesObject[uuidStringWithoutCurlyBraces(node->getUUID())] = messagesStats;
+    });
+
+    statsObject["messages"] = messagesObject;
+    ThreadedAssignment::addPacketStatsAndSendStatsPacket(statsObject);
+}
+
+void MessagesMixer::run() {
+    ThreadedAssignment::commonInit(MESSAGES_MIXER_LOGGING_NAME, NodeType::MessagesMixer);
+
+    auto nodeList = DependencyManager::get<NodeList>();
+    nodeList->addNodeTypeToInterestSet(NodeType::Agent);
+
+    // wait until we have the domain-server settings, otherwise we bail
+    DomainHandler& domainHandler = nodeList->getDomainHandler();
+
+    qDebug() << "Waiting for domain settings from domain-server.";
+
+    // block until we get the settingsRequestComplete signal
+    QEventLoop loop;
+    connect(&domainHandler, &DomainHandler::settingsReceived, &loop, &QEventLoop::quit);
+    connect(&domainHandler, &DomainHandler::settingsReceiveFail, &loop, &QEventLoop::quit);
+    domainHandler.requestDomainSettings();
+    loop.exec();
+
+    if (domainHandler.getSettingsObject().isEmpty()) {
+        qDebug() << "Failed to retreive settings object from domain-server. Bailing on assignment.";
+        setFinished(true);
+        return;
+    }
+
+    // parse the settings to pull out the values we need
+    parseDomainServerSettings(domainHandler.getSettingsObject());
+}
+
+void MessagesMixer::parseDomainServerSettings(const QJsonObject& domainSettings) {
+    // TODO - if we want options, parse them here...
+    const QString MESSAGES_MIXER_SETTINGS_KEY = "messages_mixer";
+}

assignment-client/src/messages/MessagesMixer.h

@@ -0,0 +1,43 @@
+//
+//  MessagesMixer.h
+//  assignment-client/src/messages
+//
+//  Created by Brad hefta-Gaub on 11/16/2015.
+//  Copyright 2015 High Fidelity, Inc.
+//
+//  The avatar mixer receives head, hand and positional data from all connected
+//  nodes, and broadcasts that data back to them, every BROADCAST_INTERVAL ms.
+//
+//  Distributed under the Apache License, Version 2.0.
+//  See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html
+//
+
+#ifndef hifi_MessagesMixer_h
+#define hifi_MessagesMixer_h
+
+#include <ThreadedAssignment.h>
+
+/// Handles assignments of type MessagesMixer - distribution of avatar data to various clients
+class MessagesMixer : public ThreadedAssignment {
+    Q_OBJECT
+public:
+    MessagesMixer(NLPacket& packet);
+    ~MessagesMixer();
+
+public slots:
+    void run();
+    void nodeKilled(SharedNodePointer killedNode);
+    void sendStatsPacket();
+
+private slots:
+    void handleMessages(QSharedPointer<NLPacketList> packetList, SharedNodePointer senderNode);
+    void handleMessagesSubscribe(QSharedPointer<NLPacketList> packetList, SharedNodePointer senderNode);
+    void handleMessagesUnsubscribe(QSharedPointer<NLPacketList> packetList, SharedNodePointer senderNode);
+
+private:
+    void parseDomainServerSettings(const QJsonObject& domainSettings);
+
+    QHash<QString,QSet<QUuid>> _channelSubscribers;
+};
+
+#endif // hifi_MessagesMixer_h

domain-server/src/DomainGatekeeper.cpp

@@ -48,7 +48,8 @@ QUuid DomainGatekeeper::assignmentUUIDForPendingAssignment(const QUuid& tempUUID
 
 const NodeSet STATICALLY_ASSIGNED_NODES = NodeSet() << NodeType::AudioMixer
     << NodeType::AvatarMixer << NodeType::EntityServer
-    << NodeType::AssetServer;
+    << NodeType::AssetServer
+    << NodeType::MessagesMixer;
 
 void DomainGatekeeper::processConnectRequestPacket(QSharedPointer<NLPacket> packet) {
     if (packet->getPayloadSize() == 0) {
@@ -66,7 +67,7 @@ void DomainGatekeeper::processConnectRequestPacket(QSharedPointer<NLPacket> pack
     }
     
     static const NodeSet VALID_NODE_TYPES {
-        NodeType::AudioMixer, NodeType::AvatarMixer, NodeType::AssetServer, NodeType::EntityServer, NodeType::Agent
+        NodeType::AudioMixer, NodeType::AvatarMixer, NodeType::AssetServer, NodeType::EntityServer, NodeType::Agent, NodeType::MessagesMixer
     };
     
     if (!VALID_NODE_TYPES.contains(nodeConnection.nodeType)) {

domain-server/src/DomainServer.cpp

@@ -554,7 +554,6 @@ void DomainServer::populateDefaultStaticAssignmentsExcludingTypes(const QSet<Ass
          defaultedType != Assignment::AllTypes;
          defaultedType =  static_cast<Assignment::Type>(static_cast<int>(defaultedType) + 1)) {
         if (!excludedTypes.contains(defaultedType)
-            && defaultedType != Assignment::UNUSED_0
             && defaultedType != Assignment::UNUSED_1
             && defaultedType != Assignment::AgentType) {
             

domain-server/src/DomainServerSettingsManager.cpp

@@ -131,7 +131,7 @@ void DomainServerSettingsManager::setupConfigMap(const QStringList& argumentList
     appSettings.setValue(JSON_SETTINGS_VERSION_KEY, _descriptionVersion);
 }
 
-QVariant DomainServerSettingsManager::valueOrDefaultValueForKeyPath(const QString &keyPath) {
+QVariant DomainServerSettingsManager::valueOrDefaultValueForKeyPath(const QString& keyPath) {
     const QVariant* foundValue = valueForKeyPath(_configMap.getMergedConfig(), keyPath);
 
     if (foundValue) {

examples/example/messagesExample.js

@@ -0,0 +1,43 @@
+var totalTime = 0;
+var unsubscribedForTime = 0;
+var subscribedForTime = 0;
+var subscribed = false;
+var SWITCH_SUBSCRIPTION_TIME = 10;
+Script.update.connect(function (deltaTime) {
+    var channel = "example";
+    totalTime += deltaTime;
+    if (!subscribed) {
+        unsubscribedForTime += deltaTime;
+    } else {
+        subscribedForTime += deltaTime;
+    }
+
+    if (totalTime > 5) {
+
+        // if we've been unsubscribed for SWITCH_SUBSCRIPTION_TIME seconds, subscribe
+        if (!subscribed && unsubscribedForTime > SWITCH_SUBSCRIPTION_TIME) {
+            print("---- subscribing ----");
+            Messages.subscribe(channel);
+            subscribed = true;
+            subscribedForTime = 0;
+        }
+
+        // if we've been subscribed for SWITCH_SUBSCRIPTION_TIME seconds, unsubscribe
+        if (subscribed && subscribedForTime > SWITCH_SUBSCRIPTION_TIME) {
+            print("---- unsubscribing ----");
+            Messages.unsubscribe(channel);
+            subscribed = false;
+            unsubscribedForTime = 0;
+        }
+
+        // Even if not subscribed, still publish
+        var message = "update() deltaTime:" + deltaTime;
+        //print(message);
+        Messages.sendMessage(channel, message);
+    }
+});
+
+
+Messages.messageReceived.connect(function (channel, message) {
+    print("message received on channel:" + channel + ", message:" + message);
+});

examples/example/messagesReceiverExample.js

@@ -0,0 +1,22 @@
+var totalTime = 0;
+var subscribed = false;
+var WAIT_FOR_SUBSCRIPTION_TIME = 10;
+function myUpdate(deltaTime) {
+    var channel = "example";
+    totalTime += deltaTime;
+
+    if (totalTime > WAIT_FOR_SUBSCRIPTION_TIME && !subscribed) {
+
+        print("---- subscribing ----");
+        Messages.subscribe(channel);
+        subscribed = true;
+        Script.update.disconnect(myUpdate);
+    }
+}
+
+Script.update.connect(myUpdate);
+
+Messages.messageReceived.connect(function (channel, message) {
+    print("message received on channel:" + channel + ", message:" + message);
+});
+

interface/src/Application.cpp

@@ -70,6 +70,7 @@
 #include <LogHandler.h>
 #include <MainWindow.h>
 #include <MessageDialog.h>
+#include <MessagesClient.h>
 #include <ModelEntityItem.h>
 #include <NetworkAccessManager.h>
 #include <NetworkingConstants.h>
@@ -339,6 +340,7 @@ bool setupEssentials(int& argc, char** argv) {
     DependencyManager::set<PathUtils>();
     DependencyManager::set<InterfaceActionFactory>();
     DependencyManager::set<AssetClient>();
+    DependencyManager::set<MessagesClient>();
     DependencyManager::set<UserInputMapper>();
     DependencyManager::set<controller::ScriptingInterface, ControllerScriptingInterface>();
     return true;
@@ -484,6 +486,14 @@ Application::Application(int& argc, char** argv, QElapsedTimer& startupTimer) :
     connect(assetThread, &QThread::started, assetClient.data(), &AssetClient::init);
     assetThread->start();
 
+    // Setup MessagesClient
+    auto messagesClient = DependencyManager::get<MessagesClient>();
+    QThread* messagesThread = new QThread;
+    messagesThread->setObjectName("Messages Client Thread");
+    messagesClient->moveToThread(messagesThread);
+    connect(messagesThread, &QThread::started, messagesClient.data(), &MessagesClient::init);
+    messagesThread->start();
+
     const DomainHandler& domainHandler = nodeList->getDomainHandler();
 
     connect(&domainHandler, SIGNAL(hostnameChanged(const QString&)), SLOT(domainChanged(const QString&)));
@@ -550,7 +560,7 @@ Application::Application(int& argc, char** argv, QElapsedTimer& startupTimer) :
 
     // tell the NodeList instance who to tell the domain server we care about
     nodeList->addSetOfNodeTypesToNodeInterestSet(NodeSet() << NodeType::AudioMixer << NodeType::AvatarMixer
-        << NodeType::EntityServer << NodeType::AssetServer);
+        << NodeType::EntityServer << NodeType::AssetServer << NodeType::MessagesMixer);
 
     // connect to the packet sent signal of the _entityEditSender
     connect(&_entityEditSender, &EntityEditPacketSender::packetSent, this, &Application::packetSent);

interface/src/avatar/Head.cpp

@@ -91,7 +91,6 @@ void Head::simulate(float deltaTime, bool isMine, bool billboard) {
     }
 
     if (isMine) {
-        MyAvatar* myAvatar = static_cast<MyAvatar*>(_owningAvatar);
         auto player = DependencyManager::get<recording::Deck>();
         // Only use face trackers when not playing back a recording.
         if (!player->isPlaying()) {

interface/src/avatar/SkeletonModel.cpp

@@ -247,7 +247,6 @@ void SkeletonModel::simulate(float deltaTime, bool fullUpdate) {
         return; // only simulate for own avatar
     }
 
-    MyAvatar* myAvatar = static_cast<MyAvatar*>(_owningAvatar);
     auto player = DependencyManager::get<recording::Deck>();
     if (player->isPlaying()) {
         return;

libraries/audio-client/src/AudioClient.cpp

@@ -909,13 +909,8 @@ void AudioClient::handleRecordedAudioInput(const QByteArray& audio) {
         // we don't have an audioPacket yet - set that up now
         _audioPacket = NLPacket::create(PacketType::MicrophoneAudioWithEcho);
     }
+
     // FIXME either discard stereo in the recording or record a stereo flag
-    const int numNetworkBytes = _isStereoInput
-        ? AudioConstants::NETWORK_FRAME_BYTES_STEREO
-        : AudioConstants::NETWORK_FRAME_BYTES_PER_CHANNEL;
-    const int numNetworkSamples = _isStereoInput
-        ? AudioConstants::NETWORK_FRAME_SAMPLES_STEREO
-        : AudioConstants::NETWORK_FRAME_SAMPLES_PER_CHANNEL;
 
     auto nodeList = DependencyManager::get<NodeList>();
     SharedNodePointer audioMixer = nodeList->soloNodeOfType(NodeType::AudioMixer);

libraries/networking/src/Assignment.cpp

@@ -30,6 +30,8 @@ Assignment::Type Assignment::typeForNodeType(NodeType_t nodeType) {
             return Assignment::EntityServerType;
         case NodeType::AssetServer:
             return Assignment::AssetServerType;
+        case NodeType::MessagesMixer:
+            return Assignment::MessagesMixerType;
         default:
             return Assignment::AllTypes;
     }
@@ -131,6 +133,8 @@ const char* Assignment::getTypeName() const {
             return "asset-server";
         case Assignment::EntityServerType:
             return "entity-server";
+        case Assignment::MessagesMixerType:
+            return "messages-mixer";
         default:
             return "unknown";
     }

libraries/networking/src/Assignment.h

@@ -30,7 +30,7 @@ public:
         AvatarMixerType = 1,
         AgentType = 2,
         AssetServerType = 3,
-        UNUSED_0 = 4,
+        MessagesMixerType = 4,
         UNUSED_1 = 5,
         EntityServerType = 6,
         AllTypes = 7

libraries/networking/src/MessagesClient.cpp

@@ -0,0 +1,108 @@
+//
+//  MessagesClient.cpp
+//  libraries/networking/src
+//
+//  Created by Brad hefta-Gaub on 11/16/2015.
+//  Copyright 2015 High Fidelity, Inc.
+//
+//  Distributed under the Apache License, Version 2.0.
+//  See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html
+//
+
+#include "MessagesClient.h"
+
+#include <cstdint>
+
+#include <QtCore/QBuffer>
+#include <QtCore/QThread>
+
+#include "NetworkLogging.h"
+#include "NodeList.h"
+#include "PacketReceiver.h"
+
+MessagesClient::MessagesClient() {
+    setCustomDeleter([](Dependency* dependency){
+        static_cast<MessagesClient*>(dependency)->deleteLater();
+    });
+    auto nodeList = DependencyManager::get<NodeList>();
+    auto& packetReceiver = nodeList->getPacketReceiver();
+    packetReceiver.registerMessageListener(PacketType::MessagesData, this, "handleMessagesPacket");
+    connect(nodeList.data(), &LimitedNodeList::nodeKilled, this, &MessagesClient::handleNodeKilled);
+}
+
+void MessagesClient::init() {
+    if (QThread::currentThread() != thread()) {
+        QMetaObject::invokeMethod(this, "init", Qt::BlockingQueuedConnection);
+    }
+}
+
+void MessagesClient::handleMessagesPacket(QSharedPointer<NLPacketList> packetList, SharedNodePointer senderNode) {
+    QByteArray packetData = packetList->getMessage();
+    QBuffer packet{ &packetData };
+    packet.open(QIODevice::ReadOnly);
+
+    quint16 channelLength;
+    packet.read(reinterpret_cast<char*>(&channelLength), sizeof(channelLength));
+    auto channelData = packet.read(channelLength);
+    QString channel = QString::fromUtf8(channelData);
+
+    quint16 messageLength;
+    packet.read(reinterpret_cast<char*>(&messageLength), sizeof(messageLength));
+    auto messageData = packet.read(messageLength);
+    QString message = QString::fromUtf8(messageData);
+
+    emit messageReceived(channel, message);
+}
+
+void MessagesClient::sendMessage(const QString& channel, const QString& message) {
+    auto nodeList = DependencyManager::get<NodeList>();
+    SharedNodePointer messagesMixer = nodeList->soloNodeOfType(NodeType::MessagesMixer);
+    
+    if (messagesMixer) {
+        auto packetList = NLPacketList::create(PacketType::MessagesData, QByteArray(), true, true);
+
+        auto channelUtf8 = channel.toUtf8();
+        quint16 channelLength = channelUtf8.length();
+        packetList->writePrimitive(channelLength);
+        packetList->write(channelUtf8);
+
+        auto messageUtf8 = message.toUtf8();
+        quint16 messageLength = messageUtf8.length();
+        packetList->writePrimitive(messageLength);
+        packetList->write(messageUtf8);
+
+        nodeList->sendPacketList(std::move(packetList), *messagesMixer);
+    }
+}
+
+// FIXME - we should keep track of the channels we are subscribed to locally, and
+// in the event that they mixer goes away and/or comes back we should automatically
+// resubscribe to those channels
+void MessagesClient::subscribe(const QString& channel) {
+    auto nodeList = DependencyManager::get<NodeList>();
+    SharedNodePointer messagesMixer = nodeList->soloNodeOfType(NodeType::MessagesMixer);
+
+    if (messagesMixer) {
+        auto packetList = NLPacketList::create(PacketType::MessagesSubscribe, QByteArray(), true, true);
+        packetList->write(channel.toUtf8());
+        nodeList->sendPacketList(std::move(packetList), *messagesMixer);
+    }
+}
+
+void MessagesClient::unsubscribe(const QString& channel) {
+    auto nodeList = DependencyManager::get<NodeList>();
+    SharedNodePointer messagesMixer = nodeList->soloNodeOfType(NodeType::MessagesMixer);
+
+    if (messagesMixer) {
+        auto packetList = NLPacketList::create(PacketType::MessagesUnsubscribe, QByteArray(), true, true);
+        packetList->write(channel.toUtf8());
+        nodeList->sendPacketList(std::move(packetList), *messagesMixer);
+    }
+}
+
+void MessagesClient::handleNodeKilled(SharedNodePointer node) {
+    if (node->getType() != NodeType::MessagesMixer) {
+        return;
+    }
+    // FIXME - do we need to do any special bookkeeping for when the messages mixer is no longer available
+}

libraries/networking/src/MessagesClient.h

@@ -0,0 +1,43 @@
+//
+//  MessagesClient.h
+//  libraries/networking/src
+//
+//  Created by Brad hefta-Gaub on 11/16/2015.
+//  Copyright 2015 High Fidelity, Inc.
+//
+//  Distributed under the Apache License, Version 2.0.
+//  See the accompanying file LICENSE or http://www.apache.org/licenses/LICENSE-2.0.html
+//
+
+
+#ifndef hifi_MessagesClient_h
+#define hifi_MessagesClient_h
+
+#include <QString>
+
+#include <DependencyManager.h>
+
+#include "LimitedNodeList.h"
+#include "NLPacket.h"
+#include "Node.h"
+
+class MessagesClient : public QObject, public Dependency {
+    Q_OBJECT
+public:
+    MessagesClient();
+    
+    Q_INVOKABLE void init();
+
+    Q_INVOKABLE void sendMessage(const QString& channel, const QString& message);
+    Q_INVOKABLE void subscribe(const QString& channel);
+    Q_INVOKABLE void unsubscribe(const QString& channel);
+
+signals:
+    void messageReceived(const QString& channel, const QString& message);
+
+private slots:
+    void handleMessagesPacket(QSharedPointer<NLPacketList> packetList, SharedNodePointer senderNode);
+    void handleNodeKilled(SharedNodePointer node);
+};
+
+#endif

libraries/networking/src/NetworkLogging.cpp

@@ -13,3 +13,4 @@
 
 Q_LOGGING_CATEGORY(networking, "hifi.networking")
 Q_LOGGING_CATEGORY(asset_client, "hifi.networking.asset_client")
+Q_LOGGING_CATEGORY(messages_client, "hifi.networking.messages_client")

libraries/networking/src/NetworkLogging.h

@@ -16,5 +16,6 @@
 
 Q_DECLARE_LOGGING_CATEGORY(networking)
 Q_DECLARE_LOGGING_CATEGORY(asset_client)
+Q_DECLARE_LOGGING_CATEGORY(messages_client)
 
 #endif // hifi_NetworkLogging_h

libraries/networking/src/Node.cpp

@@ -32,6 +32,7 @@ void NodeType::init() {
     TypeNameHash.insert(NodeType::Agent, "Agent");
     TypeNameHash.insert(NodeType::AudioMixer, "Audio Mixer");
     TypeNameHash.insert(NodeType::AvatarMixer, "Avatar Mixer");
+    TypeNameHash.insert(NodeType::MessagesMixer, "Messages Mixer");
     TypeNameHash.insert(NodeType::AssetServer, "Asset Server");
     TypeNameHash.insert(NodeType::Unassigned, "Unassigned");
 }

libraries/networking/src/NodeType.h

@@ -23,6 +23,7 @@ namespace NodeType {
     const NodeType_t AudioMixer = 'M';
     const NodeType_t AvatarMixer = 'W';
     const NodeType_t AssetServer = 'A';
+    const NodeType_t MessagesMixer = 'm';
     const NodeType_t Unassigned = 1;
 
     void init();

libraries/networking/src/PacketReceiver.cpp

@@ -97,7 +97,7 @@ void PacketReceiver::registerDirectListenerForTypes(PacketTypeList types,
 bool PacketReceiver::registerMessageListener(PacketType type, QObject* listener, const char* slot) {
     Q_ASSERT_X(listener, "PacketReceiver::registerMessageListener", "No object to register");
     Q_ASSERT_X(slot, "PacketReceiver::registerMessageListener", "No slot to register");
-    
+
     QMetaMethod matchingMethod = matchingMethodForListener(type, listener, slot);
 
     if (matchingMethod.isValid()) {
@@ -110,8 +110,12 @@ bool PacketReceiver::registerMessageListener(PacketType type, QObject* listener,
 
         // add the mapping
         _packetListListenerMap[type] = ObjectMethodPair(QPointer<QObject>(listener), matchingMethod);
+
+        qCDebug(networking) << "Registering a packet listener for packet list type" << type;
+
         return true;
     } else {
+        qCWarning(networking) << "FAILED to Register a packet listener for packet list type" << type;
         return false;
     }
 }
@@ -352,7 +356,7 @@ void PacketReceiver::handleVerifiedPacketList(std::unique_ptr<udt::PacketList> p
         }
         
     } else if (it == _packetListListenerMap.end()) {
-        qCWarning(networking) << "No listener found for packet type" << nlPacketList->getType();
+        qCWarning(networking) << "No listener found for packet list type" << nlPacketList->getType();
         
         // insert a dummy listener so we don't print this again
         _packetListListenerMap.insert(nlPacketList->getType(), { nullptr, QMetaMethod() });

libraries/networking/src/udt/PacketHeaders.h

@@ -87,7 +87,10 @@ public:
         AssetGetInfo,
         AssetGetInfoReply,
         DomainDisconnectRequest,
-        DomainServerRemovedNode
+        DomainServerRemovedNode,
+        MessagesData,
+        MessagesSubscribe,
+        MessagesUnsubscribe
     };
 };
 

libraries/recording/src/recording/impl/BufferClip.h

@@ -26,7 +26,8 @@ public:
 private:
     virtual FrameConstPointer readFrame(size_t index) const override;
     QString _name { QUuid().toString() };
-    mutable size_t _frameIndex { 0 };
+    
+    //mutable size_t _frameIndex { 0 }; // FIXME - not in use
 };
 
 }

libraries/script-engine/src/ScriptEngine.cpp

@@ -24,6 +24,7 @@
 #include <AudioEffectOptions.h>
 #include <AvatarData.h>
 #include <EntityScriptingInterface.h>
+#include <MessagesClient.h>
 #include <NetworkAccessManager.h>
 #include <NodeList.h>
 #include <udt/PacketHeaders.h>
@@ -366,6 +367,7 @@ void ScriptEngine::init() {
     registerGlobalObject("Vec3", &_vec3Library);
     registerGlobalObject("Uuid", &_uuidLibrary);
     registerGlobalObject("AnimationCache", DependencyManager::get<AnimationCache>().data());
+    registerGlobalObject("Messages", DependencyManager::get<MessagesClient>().data());
     qScriptRegisterMetaType(this, animVarMapToScriptValue, animVarMapFromScriptValue);
     qScriptRegisterMetaType(this, resultHandlerToScriptValue, resultHandlerFromScriptValue);
 

libraries/shared/src/shared/JSONHelpers.cpp

@@ -27,7 +27,6 @@ QJsonValue glmToJson(const T& t) {
 
 template <typename T> 
 T glmFromJson(const QJsonValue& json) {
-    static const T DEFAULT_VALUE = T();
     T result;
     if (json.isArray()) {
         QJsonArray array = json.toArray();