# 
# patch "ChangeLog"
#  from [ca7af199e3ec7cbe2e7333e3a4c107ed23161796]
#    to [7d63b013470e09c70dd05b9da8f899ec63ed2201]
# 
# patch "netsync.cc"
#  from [356f622b68dcfef9265dad15a8c988f2ea62741f]
#    to [0aef99065fae04d90bbf3ae8bcf2df03097539cd]
# 
# patch "netxx_pipe.hh"
#  from [7f13e757f0721fccd410751bdcc10f2d604de2a3]
#    to [74bdced617276bc40eca3e34f8be2b6ca8d390a9]
# 
========================================================================
--- ChangeLog	ca7af199e3ec7cbe2e7333e3a4c107ed23161796
+++ ChangeLog	7d63b013470e09c70dd05b9da8f899ec63ed2201
@@ -1,3 +1,17 @@
+2005-08-13  Christof Petig <address@hidden>
+
+	use portable PipeStream implementation to reimplement changes on the
+	ssh branch
+
+2004-06-18  Christof Petig <address@hidden>
+
+	* netsync.cc:
+	(struct session): use PipeStream, pair of sockets
+	(find_wordend,parse_ssh_url): helper functions to parse an URL
+	(call_server): recognize ssh: and file: URLs
+	(serve_stdio): variant of serve_connections for stdio
+	(run_netsync_protocol): call serve_stdio if address is -
+
 2005-07-26  Nathaniel Smith  <address@hidden>
 
 	* change_set.cc (dump): Add state_renumbering dumper.
========================================================================
--- netsync.cc	356f622b68dcfef9265dad15a8c988f2ea62741f
+++ netsync.cc	0aef99065fae04d90bbf3ae8bcf2df03097539cd
@@ -45,6 +45,7 @@
 #include "netxx/stream.h"
 #include "netxx/streamserver.h"
 #include "netxx/timeout.h"
+#include "netxx_pipe.h"
 
 //
 // this is the "new" network synchronization (netsync) system in
@@ -234,8 +235,8 @@
   app_state & app;
 
   string peer_id;
-  Netxx::socket_type fd;
-  Netxx::Stream str;  
+//  Netxx::socket_type fd;
+  shared_ptr<Netxx::StreamBase> str;  
 
   string inbuf; 
   // deque of pair<string data, size_t cur_pos>
@@ -292,8 +293,7 @@
           utf8 const & our_exclude_pattern,
           app_state & app,
           string const & peer,
-          Netxx::socket_type sock, 
-          Netxx::Timeout const & to);
+          shared_ptr<Netxx::StreamBase> sock);
 
   virtual ~session();
   
@@ -458,8 +458,7 @@
                  utf8 const & our_exclude_pattern,
                  app_state & app,
                  string const & peer,
-                 Netxx::socket_type sock, 
-                 Netxx::Timeout const & to) : 
+                 shared_ptr<Netxx::StreamBase> sock) : 
   role(role),
   voice(voice),
   our_include_pattern(our_include_pattern),
@@ -467,8 +466,7 @@
   our_matcher(our_include_pattern, our_exclude_pattern),
   app(app),
   peer_id(peer),
-  fd(sock),
-  str(sock, to),
+  str(sock),
   inbuf(""),
   outbuf_size(0),
   armed(false),
@@ -1327,10 +1325,10 @@
 {
   I(inbuf.size() < constants::netcmd_maxsz);
   char tmp[constants::bufsz];
-  Netxx::signed_size_type count = str.read(tmp, sizeof(tmp));
+  Netxx::signed_size_type count = str->read(tmp, sizeof(tmp));
   if (count > 0)
     {
-      L(F("read %d bytes from fd %d (peer %s)\n") % count % fd % peer_id);
+      L(F("read %d bytes from fd %d (peer %s)\n") % count % str->get_socketfd() % peer_id);
       if (encountered_error)
         {
           L(F("in error unwind mode, so throwing them into the bit bucket\n"));
@@ -1351,7 +1349,7 @@
 {
   I(!outbuf.empty());    
   size_t writelen = outbuf.front().first.size() - outbuf.front().second;
-  Netxx::signed_size_type count = str.write(outbuf.front().first.data() + outbuf.front().second, 
+  Netxx::signed_size_type count = str->write(outbuf.front().first.data() + outbuf.front().second, 
                                             std::min(writelen,
                                             constants::bufsz));
   if (count > 0)
@@ -1366,7 +1364,7 @@
           outbuf.front().second += count;
         }
       L(F("wrote %d bytes to fd %d (peer %s)\n")
-        % count % fd % peer_id);
+        % count % str->get_socketfd() % peer_id);
       mark_recent_io();
       if (byte_out_ticker.get() != NULL)
         (*byte_out_ticker) += count;
@@ -3215,7 +3213,42 @@
     }
 }
 
+static std::string::size_type find_wordend(const std::string &address, 
+        std::string::size_type begin, std::string::size_type end, 
+        const std::string &breaks)
+{  for (;begin<end;++begin)
+   {  for (std::string::const_iterator i=breaks.begin();i!=breaks.end();++i)
+         if (address[begin]==*i) return begin;
+   }
+   return begin;2
+}
 
+static bool parse_ssh_url(const std::string &address,std::string &host,
+                    std::string &user,std::string &port,std::string &dbpath)
+{  std::string::size_type wordbegin=0,end=address.size(),wordend=std::string::npos;
+   if (wordbegin+2<=end && address.substr(wordbegin,2)=="//") wordbegin+=2;
+   wordend=find_wordend(address,wordbegin,end,"@:/");
+   if (wordend==end) return false;
+   if (address[wordend]=='@') 
+   {  user=address.substr(wordbegin,wordend-wordbegin);
+      wordbegin=wordend+1;
+      wordend=find_wordend(address,wordbegin,end,":/");
+      if (wordend==end) return false;
+   }
+   if (address[wordend]==':') 
+   {  host=address.substr(wordbegin,wordend-wordbegin);
+      wordbegin=wordend+1;
+      wordend=find_wordend(address,wordbegin,end,"/");
+      if (wordend==end) return false;
+   }
+   if (address[wordend]!='/') return false;
+   if (wordbegin==wordend) return false; // empty port/host
+   if (host.empty()) host=address.substr(wordbegin,wordend-wordbegin);
+   else port=address.substr(wordbegin,wordend-wordbegin);
+   dbpath=address.substr(wordend); // with leading '/' !
+   return true;
+}
+
 static void 
 call_server(protocol_role role,
             utf8 const & include_pattern,
@@ -3231,9 +3264,86 @@
   // FIXME: split into labels and convert to ace here.
 
   P(F("connecting to %s\n") % address());
-  Netxx::Stream server(address().c_str(), default_port, timeout);
+  shared_ptr<Netxx::StreamBase> server;
+  if (address().substr(0,5)=="file:")
+  {  int fd1[2],fd2[2];
+     pid_t child=pipe_and_fork(fd1,fd2);
+     if (child<0)
+     {  L(F("pipe/fork failed"));
+        return;
+     }
+     else if (!child)
+     {  std::string db_path=address().substr(5);
+        const unsigned newsize=64;
+        const char *newargv[newsize];
+        unsigned newargc=0;
+        newargv[newargc++]="monotone";
+        if (global_sanity.debug) newargv[newargc++]="--debug";
+        newargv[newargc++]="--db";
+        newargv[newargc++]=db_path.c_str();
+        newargv[newargc++]="--";
+        newargv[newargc++]="serve";
+        newargv[newargc++]="-";
+        for (unsigned i=0; i<newsize-newargc-2 && i<collections.size(); ++i)
+          newargv[newargc++]=collections[i]().c_str();
+        newargv[newargc]=0;
+        
+        if (global_sanity.debug) 
+           dup2(open("monotone-server.log",O_WRONLY|O_CREAT|O_NOCTTY|O_APPEND,0666),2);
+        execvp(newargv[0],(char*const*)newargv);
+        perror(newargv[0]);
+        exit(errno);
+     }
+     server=new Netxx::PipeStream(fd1[0],fd2[1],timeout);
+  }
+  else if (address().substr(0,4)=="ssh:")
+  {  int fd1[2],fd2[2];
+     std::string user,host,port,db_path;
+     if (!parse_ssh_url(address().substr(4),host,user,port,db_path))
+     {  L(F("url %s is not of form ssh:[//address@hidden:port/dbpath\n") % address());
+        return;
+     }
+     pid_t child=pipe_and_fork(fd1,fd2);
+     if (child<0)
+     {  L(F("pipe/fork failed"));
+        return;
+     }
+     else if (!child)
+     {  const unsigned newsize=64;
+        const char *newargv[newsize];
+        unsigned newargc=0;
+        newargv[newargc++]="ssh";
+        if (!port.empty()) 
+        {  newargv[newargc++]="-p";
+           newargv[newargc++]=port.c_str();
+        }
+        if (!user.empty()) 
+        {  newargv[newargc++]="-l";
+           newargv[newargc++]=user.c_str();
+        }
+        newargv[newargc++]=host.c_str();
+        newargv[newargc++]="monotone";
+        newargv[newargc++]="--db";
+        newargv[newargc++]=db_path.c_str();
+        newargv[newargc++]="--";
+        newargv[newargc++]="serve";
+        newargv[newargc++]="-";
+        for (unsigned i=0; i<newsize-newargc-2 && i<collections.size(); ++i)
+          newargv[newargc++]=collections[i]().c_str();
+        newargv[newargc]=0;
+        
+        if (global_sanity.debug) 
+           dup2(open("monotone-server.log",O_WRONLY|O_CREAT|O_NOCTTY|O_APPEND,0666),2);
+        execvp(newargv[0],(char*const*)newargv);
+        perror(newargv[0]);
+        exit(errno);
+     }
+     server=new PipeStream(fd1[0],fd2[1],timeout);
+  }
+  else server=new Netxx::Stream(address().c_str(), default_port, timeout); 
+//  Netxx::Stream server(address().c_str(), default_port, timeout);
   session sess(role, client_voice, include_pattern, exclude_pattern,
-               app, address(), server.get_socketfd(), timeout);
+               app, address(), server);
   
   while (true)
     {       
@@ -3378,11 +3488,12 @@
   else
     {
       P(F("accepted new client connection from %s\n") % client);
+      shared_ptr<Netxx::StreamBase> str(new Netxx::Stream(client.get_socketfd(), 
+                                                          timeout));
       shared_ptr<session> sess(new session(role, server_voice,
                                            include_pattern, exclude_pattern,
                                            app,
-                                           lexical_cast<string>(client), 
-                                           client.get_socketfd(), timeout));
+                                           lexical_cast<string>(client), str));
       sess->begin_service();
       sessions.insert(make_pair(client.get_socketfd(), sess));
     }
@@ -3583,7 +3694,87 @@
     }
 }
 
+static void 
+serve_single_connection(//protocol_role role,
+                  shared_ptr<session> sess,
+                  unsigned long timeout_seconds)
+{
+  Netxx::PipeCompatibleProbe probe;  
 
+  Netxx::Timeout 
+    forever, 
+    timeout(static_cast<long>(timeout_seconds)), 
+    instant(0,1);
+
+  P(F("beginning service on %s\n") % sess->peer_id);
+
+//  Netxx::StreamServer server(addr, timeout);
+  sess->begin_service();
+  
+  map<Netxx::socket_type, shared_ptr<session> > sessions;
+  set<Netxx::socket_type> armed_sessions;
+  
+  sessions[sess->str.get_socketfd()]=sess;
+  PipeStream *pipe=dynamic_cast<PipeStream*>(&*sess->str);
+  if (pipe) sessions[pipe->get_writefd()]=sess;
+  
+  // no addr, no server
+  
+//  bool live_p = true;
+  while (!sessions.empty())
+    {      
+      probe.clear();
+      armed_sessions.clear();
+
+      arm_sessions_and_calculate_probe(probe, sessions, armed_sessions);
+
+      L(F("i/o probe with %d armed\n") % armed_sessions.size());      
+      Netxx::Probe::result_type res = probe.ready(/*sessions.empty() ? forever */
+                                           : (armed_sessions.empty() ? timeout 
+                                              : instant));
+      Netxx::Probe::ready_type event = res.second;
+      Netxx::socket_type fd = res.first;
+      
+      if (fd == -1)
+        {
+          if (armed_sessions.empty()) 
+            L(F("timed out waiting for I/O (listening on %s : %d)\n") 
+              % addr.get_name() % addr.get_port());
+        }
+      
+      // an existing session woke up
+      else
+        {
+          map<Netxx::socket_type, shared_ptr<session> >::iterator i;
+          i = sessions.find(fd);
+          if (i == sessions.end())
+            {
+              L(F("got woken up for action on unknown fd %d\n") % fd);
+            }
+          else
+            {
+              shared_ptr<session> sess = i->second;
+              bool live_p = true;
+
+              if (event & Netxx::Probe::ready_read)
+                handle_read_available(fd, sess, sessions, armed_sessions, live_p);
+                
+              if (live_p && (event & Netxx::Probe::ready_write))
+                handle_write_available(fd, sess, sessions, live_p);
+                
+              if (live_p && (event & Netxx::Probe::ready_oobd))
+                {
+                  P(F("got some OOB data on fd %d (peer %s), disconnecting\n") 
+                    % fd % sess->peer_id);
+                  sessions.erase(i);
+                }
+            }
+        }
+      process_armed_sessions(sessions, armed_sessions);
+      reap_dead_sessions(sessions, timeout_seconds);
+    }
+}
+
 /////////////////////////////////////////////////
 //
 // layer 4: monotone interface layer
@@ -3765,7 +3956,15 @@
       start_platform_netsync();
       if (voice == server_voice)
         {
-          serve_connections(role, include_pattern, exclude_pattern, app,
+          if (addr==utf8("-"))
+          { shared_ptr<Netxx::PipeStream> str(new PipeStream(0,1));
+            shared_ptr<session> sess(new session(role, server_voice, 
+                                                 include_pattern, exclude_pattern,
+                                                 app, "stdio", str));
+            serve_single_connection(sess,constants::netsync_timeout_seconds)
+          }
+          else
+            serve_connections(role, include_pattern, exclude_pattern, app,
                             addr, static_cast<Netxx::port_type>(constants::netsync_default_port), 
                             static_cast<unsigned long>(constants::netsync_timeout_seconds), 
                             static_cast<unsigned long>(constants::netsync_connection_limit));
========================================================================
--- netxx_pipe.hh	7f13e757f0721fccd410751bdcc10f2d604de2a3
+++ netxx_pipe.hh	74bdced617276bc40eca3e34f8be2b6ca8d390a9
@@ -29,7 +29,7 @@
     friend class PipeCompatibleProbe;
 #endif
 public:
-    explicit PipeStream (int readfd, int writefd);
+    explicit PipeStream (int readfd, int writefd); // Timeout?
     explicit PipeStream (const std::string &cmd, const std::vector<std::string> &args);
     virtual signed_size_type read (void *buffer, size_type length);
     virtual signed_size_type write (const void *buffer, size_type length);