Loading...
Searching...
No Matches
juce_InterprocessConnection.cpp
/*
==============================================================================
This file is part of the JUCE library.
Copyright (c) 2022 - Raw Material Software Limited
JUCE is an open source library subject to commercial or open-source
licensing.
The code included in this file is provided under the terms of the ISC license
http://www.isc.org/downloads/software-support-policy/isc-license. Permission
To use, copy, modify, and/or distribute this software for any purpose with or
without fee is hereby granted provided that the above copyright notice and
this permission notice appear in all copies.
JUCE IS PROVIDED "AS IS" WITHOUT ANY WARRANTY, AND ALL WARRANTIES, WHETHER
EXPRESSED OR IMPLIED, INCLUDING MERCHANTABILITY AND FITNESS FOR PURPOSE, ARE
DISCLAIMED.
==============================================================================
*/
namespace juce
{
struct InterprocessConnection::ConnectionThread final : public Thread
{
ConnectionThread (InterprocessConnection& c) : Thread ("JUCE IPC"), owner (c) {}
void run() override { owner.runThread(); }
InterprocessConnection& owner;
JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (ConnectionThread)
};
class SafeActionImpl
{
public:
explicit SafeActionImpl (InterprocessConnection& p)
: ref (p) {}
template <typename Fn>
void ifSafe (Fn&& fn)
{
const ScopedLock lock (mutex);
if (safe)
fn (ref);
}
void setSafe (bool s)
{
const ScopedLock lock (mutex);
safe = s;
}
bool isSafe()
{
const ScopedLock lock (mutex);
return safe;
}
private:
CriticalSection mutex;
InterprocessConnection& ref;
bool safe = false;
};
class InterprocessConnection::SafeAction final : public SafeActionImpl
{
using SafeActionImpl::SafeActionImpl;
};
//==============================================================================
InterprocessConnection::InterprocessConnection (bool callbacksOnMessageThread, uint32 magicMessageHeaderNumber)
: useMessageThread (callbacksOnMessageThread),
magicMessageHeader (magicMessageHeaderNumber),
safeAction (std::make_shared<SafeAction> (*this))
{
thread.reset (new ConnectionThread (*this));
}
InterprocessConnection::~InterprocessConnection()
{
// You *must* call `disconnect` in the destructor of your derived class to ensure
// that any pending messages are not delivered. If the messages were delivered after
// destroying the derived class, we'd end up calling the pure virtual implementations
// of `messageReceived`, `connectionMade` and `connectionLost` which is definitely
// not a good idea!
jassert (! safeAction->isSafe());
callbackConnectionState = false;
disconnect (4000, Notify::no);
thread.reset();
}
//==============================================================================
bool InterprocessConnection::connectToSocket (const String& hostName,
int portNumber, int timeOutMillisecs)
{
disconnect();
auto s = std::make_unique<StreamingSocket>();
if (s->connect (hostName, portNumber, timeOutMillisecs))
{
const ScopedWriteLock sl (pipeAndSocketLock);
initialiseWithSocket (std::move (s));
return true;
}
return false;
}
bool InterprocessConnection::connectToPipe (const String& pipeName, int timeoutMs)
{
disconnect();
auto newPipe = std::make_unique<NamedPipe>();
if (newPipe->openExisting (pipeName))
{
const ScopedWriteLock sl (pipeAndSocketLock);
pipeReceiveMessageTimeout = timeoutMs;
initialiseWithPipe (std::move (newPipe));
return true;
}
return false;
}
bool InterprocessConnection::createPipe (const String& pipeName, int timeoutMs, bool mustNotExist)
{
disconnect();
auto newPipe = std::make_unique<NamedPipe>();
if (newPipe->createNewPipe (pipeName, mustNotExist))
{
const ScopedWriteLock sl (pipeAndSocketLock);
pipeReceiveMessageTimeout = timeoutMs;
initialiseWithPipe (std::move (newPipe));
return true;
}
return false;
}
void InterprocessConnection::disconnect (int timeoutMs, Notify notify)
{
thread->signalThreadShouldExit();
{
const ScopedReadLock sl (pipeAndSocketLock);
if (socket != nullptr) socket->close();
if (pipe != nullptr) pipe->close();
}
thread->stopThread (timeoutMs);
deletePipeAndSocket();
if (notify == Notify::yes)
connectionLostInt();
callbackConnectionState = false;
safeAction->setSafe (false);
}
void InterprocessConnection::deletePipeAndSocket()
{
const ScopedWriteLock sl (pipeAndSocketLock);
socket.reset();
pipe.reset();
}
bool InterprocessConnection::isConnected() const
{
const ScopedReadLock sl (pipeAndSocketLock);
return ((socket != nullptr && socket->isConnected())
|| (pipe != nullptr && pipe->isOpen()))
&& threadIsRunning;
}
String InterprocessConnection::getConnectedHostName() const
{
{
const ScopedReadLock sl (pipeAndSocketLock);
if (pipe == nullptr && socket == nullptr)
return {};
if (socket != nullptr && ! socket->isLocal())
return socket->getHostName();
}
return IPAddress::local().toString();
}
//==============================================================================
bool InterprocessConnection::sendMessage (const MemoryBlock& message)
{
uint32 messageHeader[2] = { ByteOrder::swapIfBigEndian (magicMessageHeader),
ByteOrder::swapIfBigEndian ((uint32) message.getSize()) };
MemoryBlock messageData (sizeof (messageHeader) + message.getSize());
messageData.copyFrom (messageHeader, 0, sizeof (messageHeader));
messageData.copyFrom (message.getData(), sizeof (messageHeader), message.getSize());
return writeData (messageData.getData(), (int) messageData.getSize()) == (int) messageData.getSize();
}
int InterprocessConnection::writeData (void* data, int dataSize)
{
const ScopedReadLock sl (pipeAndSocketLock);
if (socket != nullptr)
return socket->write (data, dataSize);
if (pipe != nullptr)
return pipe->write (data, dataSize, pipeReceiveMessageTimeout);
return 0;
}
//==============================================================================
void InterprocessConnection::initialise()
{
safeAction->setSafe (true);
threadIsRunning = true;
connectionMadeInt();
thread->startThread();
}
void InterprocessConnection::initialiseWithSocket (std::unique_ptr<StreamingSocket> newSocket)
{
jassert (socket == nullptr && pipe == nullptr);
socket = std::move (newSocket);
initialise();
}
void InterprocessConnection::initialiseWithPipe (std::unique_ptr<NamedPipe> newPipe)
{
jassert (socket == nullptr && pipe == nullptr);
pipe = std::move (newPipe);
initialise();
}
//==============================================================================
struct ConnectionStateMessage final : public MessageManager::MessageBase
{
ConnectionStateMessage (std::shared_ptr<SafeActionImpl> ipc, bool connected) noexcept
: safeAction (ipc), connectionMade (connected)
{}
void messageCallback() override
{
safeAction->ifSafe ([this] (InterprocessConnection& owner)
{
if (connectionMade)
owner.connectionMade();
else
owner.connectionLost();
});
}
std::shared_ptr<SafeActionImpl> safeAction;
bool connectionMade;
JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (ConnectionStateMessage)
};
void InterprocessConnection::connectionMadeInt()
{
if (! callbackConnectionState)
{
callbackConnectionState = true;
if (useMessageThread)
(new ConnectionStateMessage (safeAction, true))->post();
else
connectionMade();
}
}
void InterprocessConnection::connectionLostInt()
{
if (callbackConnectionState)
{
callbackConnectionState = false;
if (useMessageThread)
(new ConnectionStateMessage (safeAction, false))->post();
else
connectionLost();
}
}
struct DataDeliveryMessage final : public Message
{
DataDeliveryMessage (std::shared_ptr<SafeActionImpl> ipc, const MemoryBlock& d)
: safeAction (ipc), data (d)
{}
void messageCallback() override
{
safeAction->ifSafe ([this] (InterprocessConnection& owner)
{
owner.messageReceived (data);
});
}
std::shared_ptr<SafeActionImpl> safeAction;
MemoryBlock data;
};
void InterprocessConnection::deliverDataInt (const MemoryBlock& data)
{
jassert (callbackConnectionState);
if (useMessageThread)
(new DataDeliveryMessage (safeAction, data))->post();
else
messageReceived (data);
}
//==============================================================================
int InterprocessConnection::readData (void* data, int num)
{
const ScopedReadLock sl (pipeAndSocketLock);
if (socket != nullptr)
return socket->read (data, num, true);
if (pipe != nullptr)
return pipe->read (data, num, pipeReceiveMessageTimeout);
jassertfalse;
return -1;
}
bool InterprocessConnection::readNextMessage()
{
uint32 messageHeader[2];
auto bytes = readData (messageHeader, sizeof (messageHeader));
if (bytes == (int) sizeof (messageHeader)
&& ByteOrder::swapIfBigEndian (messageHeader[0]) == magicMessageHeader)
{
auto bytesInMessage = (int) ByteOrder::swapIfBigEndian (messageHeader[1]);
if (bytesInMessage > 0)
{
MemoryBlock messageData ((size_t) bytesInMessage, true);
int bytesRead = 0;
while (bytesInMessage > 0)
{
if (thread->threadShouldExit())
return false;
auto numThisTime = jmin (bytesInMessage, 65536);
auto bytesIn = readData (addBytesToPointer (messageData.getData(), bytesRead), numThisTime);
if (bytesIn <= 0)
break;
bytesRead += bytesIn;
bytesInMessage -= bytesIn;
}
if (bytesRead >= 0)
deliverDataInt (messageData);
}
return true;
}
if (bytes < 0)
{
if (socket != nullptr)
deletePipeAndSocket();
connectionLostInt();
}
return false;
}
void InterprocessConnection::runThread()
{
while (! thread->threadShouldExit())
{
if (socket != nullptr)
{
auto ready = socket->waitUntilReady (true, 100);
if (ready < 0)
{
deletePipeAndSocket();
connectionLostInt();
break;
}
if (ready == 0)
{
thread->wait (1);
continue;
}
}
else if (pipe != nullptr)
{
if (! pipe->isOpen())
{
deletePipeAndSocket();
connectionLostInt();
break;
}
}
else
{
break;
}
if (thread->threadShouldExit() || ! readNextMessage())
break;
}
threadIsRunning = false;
}
} // namespace juce
juce::ByteOrder::swapIfBigEndian
static Type swapIfBigEndian(Type value) noexcept
Definition juce_ByteOrder.h:63
juce::IPAddress::local
static IPAddress local(bool IPv6=false) noexcept
Definition juce_IPAddress.cpp:243
juce::IPAddress::toString
String toString() const
Definition juce_IPAddress.cpp:177
juce::InterprocessConnection::connectionMade
virtual void connectionMade()=0
juce::InterprocessConnection::messageReceived
virtual void messageReceived(const MemoryBlock &message)=0
juce::InterprocessConnection::disconnect
void disconnect(int timeoutMs=-1, Notify notify=Notify::yes)
Definition juce_InterprocessConnection.cpp:148
juce::InterprocessConnection::getConnectedHostName
String getConnectedHostName() const
Definition juce_InterprocessConnection.cpp:184
juce::InterprocessConnection::InterprocessConnection
InterprocessConnection(bool callbacksOnMessageThread=true, uint32 magicMessageHeaderNumber=0xf2b49e2c)
Definition juce_InterprocessConnection.cpp:74
juce::InterprocessConnection::createPipe
bool createPipe(const String &pipeName, int pipeReceiveMessageTimeoutMs, bool mustNotExist=false)
Definition juce_InterprocessConnection.cpp:131
juce::InterprocessConnection::connectToPipe
bool connectToPipe(const String &pipeName, int pipeReceiveMessageTimeoutMs)
Definition juce_InterprocessConnection.cpp:114
juce::InterprocessConnection::isConnected
bool isConnected() const
Definition juce_InterprocessConnection.cpp:175
juce::InterprocessConnection::~InterprocessConnection
virtual ~InterprocessConnection()
Definition juce_InterprocessConnection.cpp:82
juce::InterprocessConnection::Notify
Notify
Definition juce_InterprocessConnection.h:124
juce::InterprocessConnection::connectionLost
virtual void connectionLost()=0
juce::InterprocessConnection::sendMessage
bool sendMessage(const MemoryBlock &message)
Definition juce_InterprocessConnection.cpp:200
juce::InterprocessConnection::connectToSocket
bool connectToSocket(const String &hostName, int portNumber, int timeOutMillisecs)
Definition juce_InterprocessConnection.cpp:97
juce::MemoryBlock
Definition juce_MemoryBlock.h:33
juce::MemoryBlock::copyFrom
void copyFrom(const void *srcData, int destinationOffset, size_t numBytes) noexcept
Definition juce_MemoryBlock.cpp:224
juce::MemoryBlock::getData
void * getData() noexcept
Definition juce_MemoryBlock.h:91
juce::MemoryBlock::getSize
size_t getSize() const noexcept
Definition juce_MemoryBlock.h:127
juce::MessageManager::MessageBase
Definition juce_MessageManager.h:182
juce::Message
Definition juce_Message.h:44
juce::ScopedReadLock
Definition juce_ScopedReadLock.h:52
juce::ScopedWriteLock
Definition juce_ScopedWriteLock.h:52
juce::String
Definition juce_String.h:53
juce::Thread
Definition juce_Thread.h:43
juce::Thread::Thread
Thread(const String &threadName, size_t threadStackSize=osDefaultStackSize)
Definition juce_Thread.cpp:26
juce::Thread::run
virtual void run()=0