cello_ipc.cpp

/*
    Copyright (c) 2023 Brett g Porter
    Permission is hereby granted, free of charge, to any person obtaining a copy
    of this software and associated documentation files (the "Software"), to deal
    in the Software without restriction, including without limitation the rights
    to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
    copies of the Software, and to permit persons to whom the Software is
    furnished to do so, subject to the following conditions:
    The above copyright notice and this permission notice shall be included in all
    copies or substantial portions of the Software.
    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
    AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
    OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
    SOFTWARE.
*/
 
#include "JuceHeader.h"
#include "cello_ipc.h"
 
namespace
{
// each end of a client connection must use this number in their
// headers. At some point it's probably worth finding a good way
// to parameterize this.
juce::uint32 CelloMagicIpcNumber { 0x000C3110 };
} // namespace
 
namespace juce
{
template <> struct VariantConverter<cello::IpcServerStatus>
{
    static cello::IpcServerStatus fromVar (const var& v) { return static_cast<cello::IpcServerStatus> (int (v)); }
 
    static var toVar (const cello::IpcServerStatus& t) { return static_cast<int> (t); }
};
} // namespace juce
 
namespace cello
{
 
IpcClient::IpcClient (Object& objectToWatch, UpdateType updateType, const juce::String& hostName, int portNum,
                      const juce::String& pipeName, int msTimeout, Object* state)
: juce::InterprocessConnection { true, CelloMagicIpcNumber }
, juce::ValueTreeSynchroniser { objectToWatch }
, UpdateQueue { objectToWatch, nullptr }
, clientProperties { objectToWatch.getType ().toString (), state }
, update { updateType }
, host { hostName }
, port { portNum }
, pipe { pipeName }
, timeout { msTimeout }
{
    // verify that the update type makes basic sense
    // need to either send or receive
    jassert ((update & UpdateType::send) || (update & UpdateType::receive));
    // ...and if we are sending a full update, we also need to be sending (in general)
    jassert (!(update & UpdateType::fullUpdateOnConnect) ||
             ((update & UpdateType::fullUpdateOnConnect) && (update & UpdateType::send)));
}
 
IpcClient::IpcClient (Object& objectToWatch, const juce::String& hostName, int portNum, int msTimeout,
                      UpdateType updateType, Object* state)
: IpcClient (objectToWatch, updateType, hostName, portNum, "", msTimeout, state)
{
    jassert (host.isNotEmpty ());
}
 
IpcClient::IpcClient (Object& objectToWatch, const juce::String& pipeName, int msTimeout, UpdateType updateType,
                      Object* state)
: IpcClient (objectToWatch, updateType, "", 0, pipeName, msTimeout, state)
{
    jassert (pipe.isNotEmpty ());
}
 
IpcClient::~IpcClient ()
{
    disconnect ();
}
 
bool IpcClient::connect (ConnectOptions options)
{
    if (host.isNotEmpty ())
    {
        // the options have no meaning for a socket connection, just try to
        // connect to the specified socket.
        return connectToSocket (host, port, timeout);
    }
 
    if (pipe.isNotEmpty ())
    {
        // else -- create and/or connect to a named pipe;
        switch (options)
        {
            case ConnectOptions::createOrFail:
                return createPipe (pipe, timeout, true);
            case ConnectOptions::mustExist:
                return connectToPipe (pipe, timeout);
            case ConnectOptions::createIfNeeded:
                return createPipe (pipe, timeout, false);
            case ConnectOptions::noOptions:
            default:
                jassertfalse;
                return false;
        }
    }
    // invalid state -- no host *or* pipe defined.
    jassertfalse;
    return false;
}
 
void IpcClient::connectionMade ()
{
    clientProperties.connected = true;
    if (update & UpdateType::fullUpdateOnConnect)
        sendFullSyncCallback ();
}
 
void IpcClient::connectionLost ()
{
    clientProperties.connected = false;
}
 
void IpcClient::messageReceived (const juce::MemoryBlock& message)
{
    if (update & UpdateType::receive)
    {
        // create an rvalue copy of the message that can be moved from.
        pushUpdate (juce::MemoryBlock { message });
        clientProperties.rxCount++;
    }
}
void IpcClient::stateChanged (const void* encodedChange, size_t encodedSize)
{
    if ((update & UpdateType::send) && (clientProperties.connected))
    {
        if (updateData != SyncData { encodedChange, encodedSize })
        {
            sendMessage ({ encodedChange, encodedSize });
            clientProperties.txCount++;
        }
    }
}
 
void IpcClient::startUpdate (void* data, size_t size)
{
    updateData = SyncData { data, size };
}
 
void IpcClient::endUpdate ()
{
    updateData = SyncData {};
}
 
//==============================================================================
 
IpcServerProperties::IpcServerProperties (const juce::String& path, Object* state)
: Object (path, state)
{
    // trigger callbacks when the status is set, whether the value changes or not.
    status.forceUpdate (true);
}
 
void IpcServerProperties::startServer (int portNum, const juce::String& address)
{
    if (running)
    {
        status = IpcServerStatus::alreadyRunning;
        return;
    }
    bindAddress = address;
    // setting the port number triggers a callback in the server to actually
    // start the server thread.
    portNumber = portNum;
}
 
void IpcServerProperties::stopServer ()
{
    if (!running)
    {
        status = IpcServerStatus::alreadyStopped;
        return;
    }
    bindAddress = "";
    // setting the port # < 0 will trigger the server to stop.
    portNumber = -1;
}
 
IpcServer::IpcServer (Object& sync, IpcClient::UpdateType updateType, const juce::String& statePath, Object* state)
: syncObject { sync }
, update { updateType }
, serverProperties { statePath, state }
{
    // the server properties will change its portNumber member to let us
    // know that we should start or stop ourselves.
    serverProperties.portNumber.onPropertyChange (
        [this] (const juce::Identifier& /*id*/)
        {
            if (serverProperties.portNumber > 0)
                startServer (serverProperties.portNumber, serverProperties.bindAddress);
            else
                stopServer ();
        });
}
 
IpcServer::~IpcServer ()
{
    // if the server is running, stop it.
    stopServer ();
    // ...and delete all of the connection objects.
    connections.clear ();
}
 
bool IpcServer::startServer (int portNumber, const juce::String& bindAddress)
{
    if (serverProperties.running)
    {
        serverProperties.status = IpcServerStatus::alreadyRunning;
        return true;
    }
 
    if (beginWaitingForSocket (portNumber, bindAddress))
    {
        serverProperties.running = true;
        serverProperties.status  = IpcServerStatus::startedOkay;
        return true;
    }
 
    serverProperties.running = false;
    serverProperties.status  = IpcServerStatus::errorStarting;
    return false;
}
 
bool IpcServer::stopServer ()
{
    if (!serverProperties.running)
    {
        serverProperties.status = IpcServerStatus::alreadyStopped;
        return true;
    }
 
    stop ();
    // the stop() method returns void, so we have no way to verify that
    // the server actually did stop!
 
    serverProperties.running = false;
    serverProperties.status  = IpcServerStatus::stoppedOkay;
    return true;
}
 
juce::InterprocessConnection* IpcServer::createConnectionObject ()
{
    // create a new IpcConnection object, and take over its ownership;
    // pass back a non-owning pointer to it so the base server class can
    // finish setting up the client connection.
    auto client { std::make_unique<IpcClient> (syncObject, "", 0, 0, update, &serverProperties) };
    juce::InterprocessConnection* connection { client.get () };
    connections.push_back (std::move (client));
    return connection;
}
 
} // namespace cello
 
#if RUN_UNIT_TESTS
#include "test/test_cello_ipc.inl"
#endif