summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/server/Engine.cpp11
-rw-r--r--src/server/Engine.hpp16
-rw-r--r--src/server/EventQueue.cpp6
-rw-r--r--src/server/EventQueue.hpp2
-rw-r--r--src/server/EventSource.hpp5
-rw-r--r--src/server/events/SetMetadata.hpp2
-rw-r--r--src/socket/SocketInterface.cpp149
-rw-r--r--src/socket/SocketInterface.hpp58
-rw-r--r--src/socket/SocketListener.cpp65
-rw-r--r--src/socket/SocketListener.hpp5
-rw-r--r--src/socket/ingen_socket_server.cpp8
-rw-r--r--src/socket/wscript3
12 files changed, 243 insertions, 87 deletions
diff --git a/src/server/Engine.cpp b/src/server/Engine.cpp
index f793ad22..77630d54 100644
--- a/src/server/Engine.cpp
+++ b/src/server/Engine.cpp
@@ -237,8 +237,15 @@ Engine::process_events(ProcessContext& context)
{
ThreadManager::assert_thread(THREAD_PROCESS);
- for (EventSources::iterator i = _event_sources.begin(); i != _event_sources.end(); ++i)
- (*i)->process(*_post_processor, context);
+ EventSources::iterator i = _event_sources.begin();
+ while (i != _event_sources.end()) {
+ EventSources::iterator next = i;
+ ++next;
+ if (!(*i)->process(*_post_processor, context)) {
+ _event_sources.erase(i);
+ }
+ i = next;
+ }
}
void
diff --git a/src/server/Engine.hpp b/src/server/Engine.hpp
index 9a5d4f76..8e5f381f 100644
--- a/src/server/Engine.hpp
+++ b/src/server/Engine.hpp
@@ -62,24 +62,18 @@ public:
virtual ~Engine();
+ // EngineBase methods
virtual bool activate();
-
virtual void deactivate();
-
virtual void quit();
-
virtual bool main_iteration();
-
- virtual void set_driver(SharedPtr<Driver> driver);
-
- virtual void add_event_source(SharedPtr<EventSource> source);
-
- virtual void process_events(ProcessContext& context);
-
virtual void register_client(const Raul::URI& uri, Interface* client);
-
virtual bool unregister_client(const Raul::URI& uri);
+ void set_driver(SharedPtr<Driver> driver);
+ void add_event_source(SharedPtr<EventSource> source);
+ void process_events(ProcessContext& context);
+
Ingen::Shared::World* world() const { return _world; }
ClientBroadcaster* broadcaster() const { return _broadcaster; }
diff --git a/src/server/EventQueue.cpp b/src/server/EventQueue.cpp
index 272273cd..585eaa34 100644
--- a/src/server/EventQueue.cpp
+++ b/src/server/EventQueue.cpp
@@ -67,13 +67,13 @@ EventQueue::event(Event* const ev)
*
* Executed events will be pushed to @a dest.
*/
-void
+bool
EventQueue::process(PostProcessor& dest, ProcessContext& context, bool limit)
{
ThreadManager::assert_thread(THREAD_PROCESS);
if (!_head.get())
- return;
+ return true;
/* Limit the maximum number of queued events to process per cycle. This
makes the process callback (more) realtime-safe by preventing being
@@ -105,6 +105,8 @@ EventQueue::process(PostProcessor& dest, ProcessContext& context, bool limit)
if (!next)
_tail = NULL;
}
+
+ return true;
}
/** Pre-process a single event */
diff --git a/src/server/EventQueue.hpp b/src/server/EventQueue.hpp
index a3567574..edd6c6b4 100644
--- a/src/server/EventQueue.hpp
+++ b/src/server/EventQueue.hpp
@@ -40,7 +40,7 @@ public:
explicit EventQueue();
virtual ~EventQueue();
- void process(PostProcessor& dest, ProcessContext& context, bool limit=true);
+ bool process(PostProcessor& dest, ProcessContext& context, bool limit=true);
inline bool unprepared_events() const { return _prepared_back.get(); }
inline bool empty() const { return !_head.get(); }
diff --git a/src/server/EventSource.hpp b/src/server/EventSource.hpp
index 4425586d..65afece4 100644
--- a/src/server/EventSource.hpp
+++ b/src/server/EventSource.hpp
@@ -35,7 +35,10 @@ class EventSource
public:
virtual ~EventSource() {}
- virtual void process(PostProcessor& dest,
+ /** Process events for a cycle.
+ * @return False iff this source is finished and should be removed.
+ */
+ virtual bool process(PostProcessor& dest,
ProcessContext& context,
bool limit = true) = 0;
};
diff --git a/src/server/events/SetMetadata.hpp b/src/server/events/SetMetadata.hpp
index f906193c..e1bc0c55 100644
--- a/src/server/events/SetMetadata.hpp
+++ b/src/server/events/SetMetadata.hpp
@@ -96,7 +96,7 @@ private:
typedef std::vector<SetPortValue*> SetEvents;
- Event* _create_event;
+ Event* _create_event;
SetEvents _set_events;
std::vector<SpecialType> _types;
std::vector<SpecialType> _remove_types;
diff --git a/src/socket/SocketInterface.cpp b/src/socket/SocketInterface.cpp
new file mode 100644
index 00000000..d9994e3e
--- /dev/null
+++ b/src/socket/SocketInterface.cpp
@@ -0,0 +1,149 @@
+/*
+ This file is part of Ingen.
+ Copyright 2007-2012 David Robillard <http://drobilla.net/>
+
+ Ingen is free software: you can redistribute it and/or modify it under the
+ terms of the GNU Affero General Public License as published by the Free
+ Software Foundation, either version 3 of the License, or any later version.
+
+ Ingen 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 Affero General Public License for details.
+
+ You should have received a copy of the GNU Affero General Public License
+ along with Ingen. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <errno.h>
+#include <sys/fcntl.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#include "ingen/Interface.hpp"
+#include "ingen/shared/World.hpp"
+#include "ingen/shared/AtomReader.hpp"
+#include "sord/sordmm.hpp"
+#include "sratom/sratom.h"
+#include "SocketInterface.hpp"
+
+#include "../server/Event.hpp"
+#include "../server/PostProcessor.hpp"
+#include "../server/ThreadManager.hpp"
+
+#define LOG(s) s << "[SocketInterface] "
+
+namespace Ingen {
+namespace Socket {
+
+SocketInterface::SocketInterface(Ingen::Shared::World& world, int conn)
+ : _world(world)
+ , _iface(*(Server::Engine*)world.local_engine().get(), *this)
+ , _event(NULL)
+ , _conn(conn)
+{
+ // Set connection to non-blocking so parser can read until EOF
+ // and not block indefinitely waiting for more input
+ fcntl(_conn, F_SETFL, fcntl(_conn, F_GETFL, 0) | O_NONBLOCK);
+
+ set_name("SocketInterface");
+ start();
+}
+
+SocketInterface::~SocketInterface()
+{
+ stop();
+ join();
+ close(_conn);
+}
+
+void
+SocketInterface::event(Server::Event* ev)
+{
+ assert(!_event);
+ ev->pre_process();
+ _event = ev;
+}
+
+bool
+SocketInterface::process(Server::PostProcessor& dest,
+ Server::ProcessContext& context,
+ bool limit)
+{
+ if (_event) {
+ _event->execute(context);
+ dest.append(_event, _event);
+ _event = NULL;
+ }
+ if (_conn == -1) {
+ return false;
+ }
+ return true;
+}
+
+void
+SocketInterface::_run()
+{
+ Thread::set_context(Server::THREAD_PRE_PROCESS);
+ while (!_exit_flag) {
+ // Set up a reader to parse the Turtle message into a model
+ Sord::World* world = _world.rdf_world();
+ SerdEnv* env = world->prefixes().c_obj();
+ SordModel* model = sord_new(world->c_obj(), SORD_SPO, false);
+ SerdReader* reader = sord_new_reader(model, env, SERD_TURTLE, NULL);
+ // Set base URI to path: so e.g. </foo/bar> will be a path
+ SordNode* base_uri = sord_new_uri(
+ world->c_obj(), (const uint8_t*)"path:");
+ serd_env_set_base_uri(env, sord_node_to_serd_node(base_uri));
+
+ LV2_URID_Map* map = &_world.lv2_uri_map()->urid_map_feature()->urid_map;
+
+ // Set up sratom to build an LV2_Atom from the model
+ Sratom* sratom = sratom_new(map);
+ SerdChunk chunk = { NULL, 0 };
+ LV2_Atom_Forge forge;
+ lv2_atom_forge_init(&forge, map);
+ lv2_atom_forge_set_sink(
+ &forge, sratom_forge_sink, sratom_forge_deref, &chunk);
+
+ // Read directly from the connection with serd
+ FILE* f = fdopen(_conn, "r");
+ if (!f) {
+ LOG(Raul::error) << "Failed to open connection " << _conn
+ << "(" << strerror(errno) << ")" << std::endl;
+ // Connection gone, exit
+ break;
+ }
+
+ serd_reader_read_file_handle(reader, f, (const uint8_t*)"(socket)");
+
+ // FIXME: Sratom needs work to be able to read resources
+ SordNode* msg_node = sord_new_blank(
+ world->c_obj(), (const uint8_t*)"genid1");
+
+ // Build an LV2_Atom at chunk.buf from the message
+ sratom_read(sratom, &forge, world->c_obj(), model, msg_node);
+
+ // Make an AtomReader to read that atom and do Ingen things
+ Shared::AtomReader ar(*_world.lv2_uri_map().get(),
+ *_world.uris().get(),
+ _world.forge(),
+ _iface);
+
+ // Call _iface methods based on atom content
+ ar.write((LV2_Atom*)chunk.buf);
+
+ // Respond and close connection
+ write(_conn, "OK", 2);
+ fclose(f);
+
+ sratom_free(sratom);
+ sord_node_free(world->c_obj(), msg_node);
+ serd_reader_free(reader);
+ sord_free(model);
+ }
+
+ _conn = -1;
+}
+
+} // namespace Ingen
+} // namespace Socket
diff --git a/src/socket/SocketInterface.hpp b/src/socket/SocketInterface.hpp
new file mode 100644
index 00000000..2a23cc64
--- /dev/null
+++ b/src/socket/SocketInterface.hpp
@@ -0,0 +1,58 @@
+/*
+ This file is part of Ingen.
+ Copyright 2007-2012 David Robillard <http://drobilla.net/>
+
+ Ingen is free software: you can redistribute it and/or modify it under the
+ terms of the GNU Affero General Public License as published by the Free
+ Software Foundation, either version 3 of the License, or any later version.
+
+ Ingen 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 Affero General Public License for details.
+
+ You should have received a copy of the GNU Affero General Public License
+ along with Ingen. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <string>
+
+#include "raul/SharedPtr.hpp"
+#include "raul/Thread.hpp"
+#include "ingen/Interface.hpp"
+
+#include "../server/EventSink.hpp"
+#include "../server/EventSource.hpp"
+#include "../server/EventWriter.hpp"
+
+namespace Ingen {
+
+namespace Shared { class World; }
+
+namespace Socket {
+
+class SocketInterface : public Raul::Thread
+ , public Server::EventSink
+ , public Server::EventSource
+{
+public:
+ SocketInterface(Shared::World& world, int conn);
+
+ void event(Server::Event* ev);
+
+ bool process(Server::PostProcessor& dest,
+ Server::ProcessContext& context,
+ bool limit = true);
+
+ ~SocketInterface();
+
+private:
+ virtual void _run();
+
+ Shared::World& _world;
+ Server::EventWriter _iface;
+ Server::Event* _event;
+ int _conn;
+};
+
+} // namespace Ingen
+} // namespace Socket
diff --git a/src/socket/SocketListener.cpp b/src/socket/SocketListener.cpp
index edd5421b..c2ff1c10 100644
--- a/src/socket/SocketListener.cpp
+++ b/src/socket/SocketListener.cpp
@@ -23,17 +23,18 @@
#include "ingen/shared/AtomReader.hpp"
#include "sord/sordmm.hpp"
#include "sratom/sratom.h"
+
+#include "../server/Engine.hpp"
#include "SocketListener.hpp"
+#include "SocketInterface.hpp"
#define LOG(s) s << "[SocketListener] "
namespace Ingen {
namespace Socket {
-SocketListener::SocketListener(Ingen::Shared::World& world,
- SharedPtr<Interface> iface)
+SocketListener::SocketListener(Ingen::Shared::World& world)
: _world(world)
- , _iface(iface)
{
// Create server socket
_sock = socket(AF_UNIX, SOCK_STREAM, 0);
@@ -88,60 +89,10 @@ SocketListener::_run()
continue;
}
- // Set connection to non-blocking so parser can read until EOF
- // and not block indefinitely waiting for more input
- fcntl(conn, F_SETFL, fcntl(conn, F_GETFL, 0) | O_NONBLOCK);
-
- // Set up a reader to parse the Turtle message into a model
- Sord::World* world = _world.rdf_world();
- SerdEnv* env = world->prefixes().c_obj();
- SordModel* model = sord_new(world->c_obj(), SORD_SPO, false);
- SerdReader* reader = sord_new_reader(model, env, SERD_TURTLE, NULL);
-
- // Set base URI to path: so e.g. </foo/bar> will be a path
- SordNode* base_uri = sord_new_uri(
- world->c_obj(), (const uint8_t*)"path:");
- serd_env_set_base_uri(env, sord_node_to_serd_node(base_uri));
-
- LV2_URID_Map* map = &_world.lv2_uri_map()->urid_map_feature()->urid_map;
-
- // Set up sratom to build an LV2_Atom from the model
- Sratom* sratom = sratom_new(map);
- SerdChunk chunk = { NULL, 0 };
- LV2_Atom_Forge forge;
- lv2_atom_forge_init(&forge, map);
- lv2_atom_forge_set_sink(
- &forge, sratom_forge_sink, sratom_forge_deref, &chunk);
-
- // Read directly from the connection with serd
- FILE* f = fdopen(conn, "r");
- serd_reader_read_file_handle(reader, f, (const uint8_t*)"(socket)");
-
- // FIXME: Sratom needs work to be able to read resources
- SordNode* msg_node = sord_new_blank(
- world->c_obj(), (const uint8_t*)"genid1");
-
- // Build an LV2_Atom at chunk.buf from the message
- sratom_read(sratom, &forge, world->c_obj(), model, msg_node);
-
- // Make an AtomReader to read that atom and do Ingen things
- Shared::AtomReader ar(*_world.lv2_uri_map().get(),
- *_world.uris().get(),
- _world.forge(),
- *_iface.get());
-
- // Call _iface methods based on atom content
- ar.write((LV2_Atom*)chunk.buf);
-
- // Respond and close connection
- write(conn, "OK", 2);
- fclose(f);
- close(conn);
-
- sratom_free(sratom);
- sord_node_free(world->c_obj(), msg_node);
- serd_reader_free(reader);
- sord_free(model);
+ // Make an new interface/thread to handle the connection
+ Server::Engine* engine = (Server::Engine*)_world.local_engine().get();
+ SharedPtr<SocketInterface> iface(new SocketInterface(_world, conn));
+ engine->add_event_source(iface);
}
}
diff --git a/src/socket/SocketListener.hpp b/src/socket/SocketListener.hpp
index 896b08b4..10e9d1d2 100644
--- a/src/socket/SocketListener.hpp
+++ b/src/socket/SocketListener.hpp
@@ -30,16 +30,13 @@ namespace Socket {
class SocketListener : public Raul::Thread
{
public:
- SocketListener(Ingen::Shared::World& world,
- SharedPtr<Interface> iface);
-
+ SocketListener(Ingen::Shared::World& world);
~SocketListener();
private:
virtual void _run();
Ingen::Shared::World& _world;
- SharedPtr<Interface> _iface;
std::string _sock_path;
int _sock;
};
diff --git a/src/socket/ingen_socket_server.cpp b/src/socket/ingen_socket_server.cpp
index 19257add..71b8ad64 100644
--- a/src/socket/ingen_socket_server.cpp
+++ b/src/socket/ingen_socket_server.cpp
@@ -29,14 +29,8 @@ using namespace Ingen;
struct IngenSocketServerModule : public Ingen::Shared::Module {
void load(Ingen::Shared::World* world) {
- Server::Engine* engine = (Server::Engine*)world->local_engine().get();
- SharedPtr<Server::EventQueue> queue(new Server::EventQueue());
- SharedPtr<Server::EventWriter> interface(
- new Server::EventWriter(*engine, *queue.get()));
listener = SharedPtr<Ingen::Socket::SocketListener>(
- new Ingen::Socket::SocketListener(*world, interface));
-
- engine->add_event_source(queue);
+ new Ingen::Socket::SocketListener(*world));
}
SharedPtr<Ingen::Socket::SocketListener> listener;
diff --git a/src/socket/wscript b/src/socket/wscript
index 6b104906..6985f377 100644
--- a/src/socket/wscript
+++ b/src/socket/wscript
@@ -4,7 +4,8 @@ from waflib.extras import autowaf as autowaf
def build(bld):
if bld.is_defined('HAVE_SOCKET'):
obj = bld(features = 'cxx cxxshlib',
- source = ['SocketListener.cpp',
+ source = ['SocketInterface.cpp',
+ 'SocketListener.cpp',
'ingen_socket_server.cpp'],
includes = ['.', '../..'],
name = 'libingen_socket_server',
generated by cgit v1.2.1 (git 2.18.0) at 2024-12-27 10:34:40 +0000