From c0e6f7ecbf77ef59e7e4948dd96109417a4994d5 Mon Sep 17 00:00:00 2001 From: David Robillard Date: Wed, 9 May 2012 02:04:05 +0000 Subject: SocketInterface => SocketReader. git-svn-id: http://svn.drobilla.net/lad/trunk/ingen@4325 a436a847-0d15-0410-975c-d299462d15a1 --- src/socket/SocketInterface.cpp | 187 ----------------------------------------- src/socket/SocketInterface.hpp | 61 -------------- src/socket/SocketListener.cpp | 4 +- src/socket/SocketReader.cpp | 187 +++++++++++++++++++++++++++++++++++++++++ src/socket/SocketReader.hpp | 68 +++++++++++++++ src/socket/wscript | 2 +- 6 files changed, 258 insertions(+), 251 deletions(-) delete mode 100644 src/socket/SocketInterface.cpp delete mode 100644 src/socket/SocketInterface.hpp create mode 100644 src/socket/SocketReader.cpp create mode 100644 src/socket/SocketReader.hpp (limited to 'src/socket') diff --git a/src/socket/SocketInterface.cpp b/src/socket/SocketInterface.cpp deleted file mode 100644 index 8b9aab66..00000000 --- a/src/socket/SocketInterface.cpp +++ /dev/null @@ -1,187 +0,0 @@ -/* - This file is part of Ingen. - Copyright 2007-2012 David Robillard - - 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 . -*/ - -#include -#include - -#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" - -#define LOG(s) s << "[SocketInterface] " - -namespace Ingen { -namespace Socket { - -SocketInterface::SocketInterface(Ingen::Shared::World& world, - Interface& iface, - int conn) - : _world(world) - , _iface(iface) - , _inserter(NULL) - , _msg_node(NULL) - , _conn(conn) -{ - set_name("SocketInterface"); - start(); -} - -SocketInterface::~SocketInterface() -{ - stop(); - join(); - close(_conn); -} - -SerdStatus -SocketInterface::set_base_uri(SocketInterface* iface, - const SerdNode* uri_node) -{ - return sord_inserter_set_base_uri(iface->_inserter, uri_node); -} - -SerdStatus -SocketInterface::set_prefix(SocketInterface* iface, - const SerdNode* name, - const SerdNode* uri_node) -{ - return sord_inserter_set_prefix(iface->_inserter, name, uri_node); -} - -SerdStatus -SocketInterface::write_statement(SocketInterface* iface, - SerdStatementFlags flags, - const SerdNode* graph, - const SerdNode* subject, - const SerdNode* predicate, - const SerdNode* object, - const SerdNode* object_datatype, - const SerdNode* object_lang) -{ - if (!iface->_msg_node) { - iface->_msg_node = sord_node_from_serd_node( - iface->_world.rdf_world()->c_obj(), iface->_env, subject, 0, 0); - } - - return sord_inserter_write_statement( - iface->_inserter, flags, graph, - subject, predicate, object, - object_datatype, object_lang); -} - -void -SocketInterface::_run() -{ - Sord::World* world = _world.rdf_world(); - LV2_URID_Map* map = &_world.lv2_uri_map()->urid_map_feature()->urid_map; - - // Use as base URI so e.g. will be a path - SordNode* base_uri = sord_new_uri( - world->c_obj(), (const uint8_t*)"path:"); - - // Set up sratom and a forge to build LV2 atoms from 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); - - // Make a model and reader to parse the next Turtle message - _env = world->prefixes().c_obj(); - SordModel* model = sord_new(world->c_obj(), SORD_SPO, false); - - _inserter = sord_inserter_new(model, _env); - - SerdReader* reader = serd_reader_new( - SERD_TURTLE, this, NULL, - (SerdBaseSink)set_base_uri, - (SerdPrefixSink)set_prefix, - (SerdStatementSink)write_statement, - NULL); - - serd_env_set_base_uri(_env, sord_node_to_serd_node(base_uri)); - - // 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 - _conn = -1; - return; - } - - serd_reader_start_stream(reader, f, (const uint8_t*)"(socket)", false); - - // Make an AtomReader to call Ingen Interface methods based on Atom - Shared::AtomReader ar(*_world.lv2_uri_map().get(), - *_world.uris().get(), - _world.forge(), - _iface); - - struct pollfd pfd; - pfd.fd = _conn; - pfd.events = POLLIN; - pfd.revents = 0; - - while (!_exit_flag) { - // Wait for input to arrive at socket - int ret = poll(&pfd, 1, -1); - if (ret == -1 || (pfd.revents & (POLLHUP | POLLNVAL))) { - break; // Hangup - } else if (!ret) { - continue; // No data, shouldn't happen - } - - // Read until the next '.' - SerdStatus st = serd_reader_read_chunk(reader); - if (st == SERD_FAILURE) { - continue; // Read nothing, e.g. just whitespace - } else if (st) { - fprintf(stderr, "Read error: %s\n", serd_strerror(st)); - } - - // Build an LV2_Atom at chunk.buf from the message - sratom_read(sratom, &forge, world->c_obj(), model, _msg_node); - - // Call _iface methods based on atom content - ar.write((LV2_Atom*)chunk.buf); - - // Respond and close connection - write(_conn, "OK", 2); - - // Reset everything for the next iteration - chunk.len = 0; - sord_node_free(world->c_obj(), _msg_node); - _msg_node = NULL; - } - - fclose(f); - sord_inserter_free(_inserter); - serd_reader_end_stream(reader); - sratom_free(sratom); - serd_reader_free(reader); - sord_free(model); - - delete this; -} - -} // namespace Ingen -} // namespace Socket diff --git a/src/socket/SocketInterface.hpp b/src/socket/SocketInterface.hpp deleted file mode 100644 index e084d964..00000000 --- a/src/socket/SocketInterface.hpp +++ /dev/null @@ -1,61 +0,0 @@ -/* - This file is part of Ingen. - Copyright 2007-2012 David Robillard - - 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 . -*/ - -#include "ingen/Interface.hpp" -#include "raul/Thread.hpp" -#include "sord/sord.h" - -namespace Ingen { - -namespace Shared { class World; } - -namespace Socket { - -class SocketInterface : public Raul::Thread -{ -public: - SocketInterface(Shared::World& world, Interface& iface, int conn); - ~SocketInterface(); - -private: - virtual void _run(); - - static SerdStatus set_base_uri(SocketInterface* iface, - const SerdNode* uri_node); - - static SerdStatus set_prefix(SocketInterface* iface, - const SerdNode* name, - const SerdNode* uri_node); - - static SerdStatus write_statement(SocketInterface* iface, - SerdStatementFlags flags, - const SerdNode* graph, - const SerdNode* subject, - const SerdNode* predicate, - const SerdNode* object, - const SerdNode* object_datatype, - const SerdNode* object_lang); - - Shared::World& _world; - Interface& _iface; - SerdEnv* _env; - SordInserter* _inserter; - SordNode* _msg_node; - int _conn; -}; - -} // namespace Ingen -} // namespace Socket diff --git a/src/socket/SocketListener.cpp b/src/socket/SocketListener.cpp index aeb50c5f..79ad11c4 100644 --- a/src/socket/SocketListener.cpp +++ b/src/socket/SocketListener.cpp @@ -27,7 +27,7 @@ #include "../server/Engine.hpp" #include "../server/EventWriter.hpp" #include "SocketListener.hpp" -#include "SocketInterface.hpp" +#include "SocketReader.hpp" #define LOG(s) s << "[SocketListener] " @@ -94,7 +94,7 @@ SocketListener::_run() // Make an new interface/thread to handle the connection Server::Engine* engine = (Server::Engine*)_world.local_engine().get(); - new SocketInterface(_world, *engine->interface(), conn); + new SocketReader(_world, *engine->interface(), conn); } } diff --git a/src/socket/SocketReader.cpp b/src/socket/SocketReader.cpp new file mode 100644 index 00000000..8b3ed7a1 --- /dev/null +++ b/src/socket/SocketReader.cpp @@ -0,0 +1,187 @@ +/* + This file is part of Ingen. + Copyright 2007-2012 David Robillard + + 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 . +*/ + +#include +#include + +#include "ingen/Interface.hpp" +#include "ingen/shared/World.hpp" +#include "ingen/shared/AtomReader.hpp" +#include "sord/sordmm.hpp" +#include "sratom/sratom.h" +#include "SocketReader.hpp" + +#define LOG(s) s << "[SocketReader] " + +namespace Ingen { +namespace Socket { + +SocketReader::SocketReader(Ingen::Shared::World& world, + Interface& iface, + int conn) + : _world(world) + , _iface(iface) + , _inserter(NULL) + , _msg_node(NULL) + , _conn(conn) +{ + set_name("SocketReader"); + start(); +} + +SocketReader::~SocketReader() +{ + stop(); + join(); + close(_conn); +} + +SerdStatus +SocketReader::set_base_uri(SocketReader* iface, + const SerdNode* uri_node) +{ + return sord_inserter_set_base_uri(iface->_inserter, uri_node); +} + +SerdStatus +SocketReader::set_prefix(SocketReader* iface, + const SerdNode* name, + const SerdNode* uri_node) +{ + return sord_inserter_set_prefix(iface->_inserter, name, uri_node); +} + +SerdStatus +SocketReader::write_statement(SocketReader* iface, + SerdStatementFlags flags, + const SerdNode* graph, + const SerdNode* subject, + const SerdNode* predicate, + const SerdNode* object, + const SerdNode* object_datatype, + const SerdNode* object_lang) +{ + if (!iface->_msg_node) { + iface->_msg_node = sord_node_from_serd_node( + iface->_world.rdf_world()->c_obj(), iface->_env, subject, 0, 0); + } + + return sord_inserter_write_statement( + iface->_inserter, flags, graph, + subject, predicate, object, + object_datatype, object_lang); +} + +void +SocketReader::_run() +{ + Sord::World* world = _world.rdf_world(); + LV2_URID_Map* map = &_world.lv2_uri_map()->urid_map_feature()->urid_map; + + // Use as base URI so e.g. will be a path + SordNode* base_uri = sord_new_uri( + world->c_obj(), (const uint8_t*)"path:"); + + // Set up sratom and a forge to build LV2 atoms from 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); + + // Make a model and reader to parse the next Turtle message + _env = world->prefixes().c_obj(); + SordModel* model = sord_new(world->c_obj(), SORD_SPO, false); + + _inserter = sord_inserter_new(model, _env); + + SerdReader* reader = serd_reader_new( + SERD_TURTLE, this, NULL, + (SerdBaseSink)set_base_uri, + (SerdPrefixSink)set_prefix, + (SerdStatementSink)write_statement, + NULL); + + serd_env_set_base_uri(_env, sord_node_to_serd_node(base_uri)); + + // 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 + _conn = -1; + return; + } + + serd_reader_start_stream(reader, f, (const uint8_t*)"(socket)", false); + + // Make an AtomReader to call Ingen Interface methods based on Atom + Shared::AtomReader ar(*_world.lv2_uri_map().get(), + *_world.uris().get(), + _world.forge(), + _iface); + + struct pollfd pfd; + pfd.fd = _conn; + pfd.events = POLLIN; + pfd.revents = 0; + + while (!_exit_flag) { + // Wait for input to arrive at socket + int ret = poll(&pfd, 1, -1); + if (ret == -1 || (pfd.revents & (POLLHUP | POLLNVAL))) { + break; // Hangup + } else if (!ret) { + continue; // No data, shouldn't happen + } + + // Read until the next '.' + SerdStatus st = serd_reader_read_chunk(reader); + if (st == SERD_FAILURE) { + continue; // Read nothing, e.g. just whitespace + } else if (st) { + fprintf(stderr, "Read error: %s\n", serd_strerror(st)); + } + + // Build an LV2_Atom at chunk.buf from the message + sratom_read(sratom, &forge, world->c_obj(), model, _msg_node); + + // Call _iface methods based on atom content + ar.write((LV2_Atom*)chunk.buf); + + // Respond and close connection + write(_conn, "OK", 2); + + // Reset everything for the next iteration + chunk.len = 0; + sord_node_free(world->c_obj(), _msg_node); + _msg_node = NULL; + } + + fclose(f); + sord_inserter_free(_inserter); + serd_reader_end_stream(reader); + sratom_free(sratom); + serd_reader_free(reader); + sord_free(model); + + delete this; +} + +} // namespace Ingen +} // namespace Socket diff --git a/src/socket/SocketReader.hpp b/src/socket/SocketReader.hpp new file mode 100644 index 00000000..141e6216 --- /dev/null +++ b/src/socket/SocketReader.hpp @@ -0,0 +1,68 @@ +/* + This file is part of Ingen. + Copyright 2007-2012 David Robillard + + 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 . +*/ + +#ifndef INGEN_SOCKET_SOCKET_READER_HPP +#define INGEN_SOCKET_SOCKET_READER_HPP + +#include "raul/Thread.hpp" +#include "sord/sord.h" + +namespace Ingen { + +namespace Shared { +class World; +class Interface; +} + +namespace Socket { + +class SocketReader : public Raul::Thread +{ +public: + SocketReader(Shared::World& world, Interface& iface, int conn); + ~SocketReader(); + +private: + virtual void _run(); + + static SerdStatus set_base_uri(SocketReader* iface, + const SerdNode* uri_node); + + static SerdStatus set_prefix(SocketReader* iface, + const SerdNode* name, + const SerdNode* uri_node); + + static SerdStatus write_statement(SocketReader* iface, + SerdStatementFlags flags, + const SerdNode* graph, + const SerdNode* subject, + const SerdNode* predicate, + const SerdNode* object, + const SerdNode* object_datatype, + const SerdNode* object_lang); + + Shared::World& _world; + Interface& _iface; + SerdEnv* _env; + SordInserter* _inserter; + SordNode* _msg_node; + int _conn; +}; + +} // namespace Ingen +} // namespace Socket + +#endif // INGEN_SOCKET_SOCKET_READER_HPP diff --git a/src/socket/wscript b/src/socket/wscript index 6985f377..e351e0e4 100644 --- a/src/socket/wscript +++ b/src/socket/wscript @@ -4,7 +4,7 @@ from waflib.extras import autowaf as autowaf def build(bld): if bld.is_defined('HAVE_SOCKET'): obj = bld(features = 'cxx cxxshlib', - source = ['SocketInterface.cpp', + source = ['SocketReader.cpp', 'SocketListener.cpp', 'ingen_socket_server.cpp'], includes = ['.', '../..'], -- cgit v1.2.1