Persistent socket interface and interactive shell.

git-svn-id: http://svn.drobilla.net/lad/trunk/ingen@4322 a436a847-0d15-0410-975c-d299462d15a1

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 <http://drobilla.net>
 #
 # 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 <errno.h>
-#include <sys/fcntl.h>
-#include <sys/socket.h>
-#include <sys/un.h>
+#include <poll.h>
 
 #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 <path:> as base URI so e.g. </foo/bar> 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. </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;
+		// 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 <string>
 
+#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: