From 4d46a232b30be99bc34e581cbc636345f77c6bc4 Mon Sep 17 00:00:00 2001 From: David Robillard Date: Wed, 9 May 2012 00:00:48 +0000 Subject: Persistent socket interface and interactive shell. git-svn-id: http://svn.drobilla.net/lad/trunk/ingen@4322 a436a847-0d15-0410-975c-d299462d15a1 --- scripts/ingenish | 82 +++++++++++-------- src/socket/SocketInterface.cpp | 179 ++++++++++++++++++++++++++++------------- src/socket/SocketInterface.hpp | 24 +++++- wscript | 2 + 4 files changed, 198 insertions(+), 89 deletions(-) diff --git a/scripts/ingenish b/scripts/ingenish index b88424ca..9e49c271 100755 --- a/scripts/ingenish +++ b/scripts/ingenish @@ -1,5 +1,5 @@ #!/usr/bin/env python -# Ingen Command-Line Interface +# Ingen Interactive Shell # Copyright 2011-2012 David Robillard # # Permission to use, copy, modify, and/or distribute this software for any @@ -15,13 +15,23 @@ # OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. import os.path +import shlex import socket import sys import time +try: + import readline +except: + pass + class Client: def __init__(self, path='/tmp/ingen.sock'): - self.path = path + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.connect(path) + + def __del__(self): + self.sock.close() def msgencode(self, msg): if sys.version_info[0] == 3: @@ -30,11 +40,8 @@ class Client: return msg def send(self, msg): - self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - self.sock.connect(self.path) self.sock.send(self.msgencode(msg)) response = self.sock.recv(1024) - self.sock.close() if response != self.msgencode('OK'): print('Error: %s' % response) return False @@ -93,7 +100,11 @@ class Client: def print_usage(): print('''Usage: ingenish COMMAND [ARGUMENT]... -A command line interface to an Ingen server. + ingenish + +A command line interface to an Ingen server. A command can be given directly +on the command line, or when run with no arguments an interactive shell is +launched. Commands: put PATH TURTLE_FRAGMENT @@ -101,6 +112,8 @@ Commands: connect TAIL_PATH HEAD_PATH disconnect TAIL_PATH HEAD_PATH delete PATH + help + exit Paths are UNIX-style paths with strict LV2 symbol components, e.g. /foo/bar_2. Turtle fragments are used verbatim as the body of blank nodes, the syntax is @@ -117,36 +130,37 @@ Example: ingenish set /tone/output 'ingen:value 0.7' ''') -def abort_if_num_args_less_than(num): - if len(sys.argv) < num: - print_usage() - sys.exit(1) - -abort_if_num_args_less_than(3) ingen = Client() -cmd = sys.argv[1] -success = 0 -if cmd == 'put': - abort_if_num_args_less_than(3) - success = ingen.put(sys.argv[2], sys.argv[3]) -elif cmd == 'set': - abort_if_num_args_less_than(3) - success = ingen.set(sys.argv[2], sys.argv[3]) -elif cmd == 'connect': - abort_if_num_args_less_than(4) - success = ingen.connect(sys.argv[2], sys.argv[3]) -elif cmd == 'disconnect': - abort_if_num_args_less_than(4) - success = ingen.disconnect(sys.argv[2], sys.argv[3]) -elif cmd == 'delete': - success = ingen.delete(sys.argv[2]) -else: - print("error: Unknown command `%s'" % cmd) - print_usage() - sys.exit(1) - -if success: +def run(cmd): + if cmd[0] == 'help': + print_usage() + elif cmd[0] == 'exit': + sys.exit(0) + elif cmd[0] == 'put' and len(cmd) == 3: + return ingen.put(cmd[1], cmd[2]) + elif cmd[0] == 'set' and len(cmd) == 3: + return ingen.set(cmd[1], cmd[2]) + elif cmd[0] == 'connect' and len(cmd) == 3: + return ingen.connect(cmd[1], cmd[2]) + elif cmd[0] == 'disconnect' and len(cmd) == 3: + return ingen.disconnect(cmd[1], cmd[2]) + elif cmd[0] == 'delete' and len(cmd) == 2: + return ingen.delete(cmd[1]) + else: + return False + +if len(sys.argv) == 1: + while True: + try: + run(shlex.split(raw_input('> '))) + except (EOFError, KeyboardInterrupt, SystemExit): + print('') + break + except: + print('error: %s' % sys.exc_info()[0]) +elif run(sys.argv[1:]): sys.exit(0) else: + print_usage() sys.exit(1) diff --git a/src/socket/SocketInterface.cpp b/src/socket/SocketInterface.cpp index d9994e3e..430f0962 100644 --- a/src/socket/SocketInterface.cpp +++ b/src/socket/SocketInterface.cpp @@ -15,9 +15,7 @@ */ #include -#include -#include -#include +#include #include "ingen/Interface.hpp" #include "ingen/shared/World.hpp" @@ -38,19 +36,18 @@ namespace Socket { SocketInterface::SocketInterface(Ingen::Shared::World& world, int conn) : _world(world) , _iface(*(Server::Engine*)world.local_engine().get(), *this) + , _inserter(NULL) + , _msg_node(NULL) , _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() { + std::cerr << "SOCKET INTERFACE EXITING" << std::endl; stop(); join(); close(_conn); @@ -59,9 +56,14 @@ SocketInterface::~SocketInterface() void SocketInterface::event(Server::Event* ev) { + if (_event) { + std::cerr << "DUAL EVENTS" << std::endl; + return; + } assert(!_event); ev->pre_process(); _event = ev; + _event->next(NULL); } bool @@ -74,74 +76,143 @@ SocketInterface::process(Server::PostProcessor& dest, dest.append(_event, _event); _event = NULL; } - if (_conn == -1) { - return false; + return (_conn != -1); +} + +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 true; + + return sord_inserter_write_statement( + iface->_inserter, flags, graph, + subject, predicate, object, + object_datatype, object_lang); } void SocketInterface::_run() { Thread::set_context(Server::THREAD_PRE_PROCESS); + + 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) { - // 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. 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; + // 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 } - 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"); + // 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); - - // 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); + 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); - fclose(f); - sratom_free(sratom); - sord_node_free(world->c_obj(), msg_node); - serd_reader_free(reader); - sord_free(model); + // 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); + _conn = -1; } diff --git a/src/socket/SocketInterface.hpp b/src/socket/SocketInterface.hpp index 2a23cc64..b8d339f8 100644 --- a/src/socket/SocketInterface.hpp +++ b/src/socket/SocketInterface.hpp @@ -16,9 +16,10 @@ #include +#include "ingen/Interface.hpp" #include "raul/SharedPtr.hpp" #include "raul/Thread.hpp" -#include "ingen/Interface.hpp" +#include "sord/sord.h" #include "../server/EventSink.hpp" #include "../server/EventSource.hpp" @@ -45,11 +46,32 @@ public: ~SocketInterface(); + SordInserter* inserter() { return _inserter; } + 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; Server::EventWriter _iface; + SerdEnv* _env; + SordInserter* _inserter; + SordNode* _msg_node; Server::Event* _event; int _conn; }; diff --git a/wscript b/wscript index 16f7ec52..a335fab0 100644 --- a/wscript +++ b/wscript @@ -65,6 +65,8 @@ def configure(conf): atleast_version='0.1.0', mandatory=True) autowaf.check_pkg(conf, 'raul', uselib_store='RAUL', atleast_version='0.8.5', mandatory=True) + autowaf.check_pkg(conf, 'serd-0', uselib_store='SERD', + atleast_version='0.15.0', mandatory=False) autowaf.check_pkg(conf, 'sord-0', uselib_store='SORD', atleast_version='0.7.0', mandatory=False) if not Options.options.no_gui: -- cgit v1.2.1