update distributor

vendor/distributor/Gemfile.lock

@@ -1,13 +1,13 @@
 PATH
   remote: .
   specs:
-    distributor (0.5.1)
+    distributor (0.6.0)
       thor (>= 0.13.6)
 
 GEM
   remote: http://rubygems.org/
   specs:
-    thor (0.15.3)
+    thor (0.16.0)
 
 PLATFORMS
   ruby

vendor/distributor/example/local.rb

@@ -44,10 +44,15 @@ def set_buffer(enable)
     client.run("bash 2>&1") do |ch|
       client.hookup ch, $stdin.dup, $stdout.dup
       client.on_close(ch) do
+        p [:closing]
         exit 0
       end
     end
 
+    client.run("ls -la") do |ch|
+      client.on_close(ch) { puts "ls closed" }
+    end
+
     # echo commands to stdout
     client.on_command do |command, data|
       $stdout.puts "received: #{command} #{data.inspect}"
@@ -74,6 +79,24 @@ def set_buffer(enable)
       end
     end
 
+    # create a unix socket to proxy through
+    unix = UNIXServer.new("/tmp/client.sock")
+
+    Thread.new do
+      loop do
+
+        # every time a connection comes to localhost:8000 on the client
+        Thread.start(unix.accept) do |unix_client|
+
+          # create a tunnel to localhost:5000 on the server
+          client.tunnel(5000) do |ch|
+            client.hookup ch, unix_client
+          end
+
+        end
+      end
+    end
+
     # turn off terminal specials
     set_buffer false
 

vendor/distributor/lib/distributor/client.rb

@@ -68,6 +68,11 @@ def run(command, &handler)
     @handlers[id] = handler
   end
 
+  def socket(path, &handler)
+    id = command("socket", "path" => path)
+    @handlers[id] = handler
+  end
+
   def tunnel(port, &handler)
     id = command("tunnel", "port" => port)
     @handlers[id] = handler

vendor/distributor/lib/distributor/server.rb

@@ -30,10 +30,14 @@ def initialize(input, output=input)
 
       dequeue_json do |data|
         case command = data["command"]
+        when "socket" then
+          path = data["path"]
+          ch = socket(path)
+          ack data["id"], "ch" => ch, "path" => path
         when "tunnel" then
           port = (data["port"] || ENV["PORT"] || 5000).to_i
           ch = tunnel(port)
-          @multiplexer.output 0, Distributor::OkJson.encode({ "id" => data["id"], "command" => "ack", "ch" => ch, "port" => port })
+          ack data["id"], "ch" => ch, "port" => port
         when "close" then
           @multiplexer.close data["ch"]
         when "run" then
@@ -46,6 +50,10 @@ def initialize(input, output=input)
     end
   end
 
+  def ack(id, options={})
+    @multiplexer.output 0, Distributor::OkJson.encode(options.merge({ "id" => id, "command" => "ack" }))
+  end
+
   def run(command)
     ch = @multiplexer.reserve
 
@@ -70,13 +78,9 @@ def run(command)
     ch
   end
 
-  def tunnel(port)
-    ch = @multiplexer.reserve
-
-    tcp = TCPSocket.new("localhost", port)
-
+  def handle_socket(ch, socket)
     # handle data incoming from process
-    @connector.handle(tcp) do |io|
+    @connector.handle(socket) do |io|
       begin
         @multiplexer.output(ch, io.readpartial(4096))
       rescue EOFError
@@ -88,12 +92,22 @@ def tunnel(port)
     # handle data incoming on the multiplexer
     @connector.handle(@multiplexer.reader(ch)) do |input_io|
       data = input_io.readpartial(4096)
-      tcp.write data
+      socket.write data
     end
 
     ch
   end
 
+  def socket(path)
+    ch = @multiplexer.reserve
+    handle_socket ch, UNIXSocket.new(path)
+  end
+
+  def tunnel(port)
+    ch = @multiplexer.reserve
+    handle_socket ch, TCPSocket.new("localhost", port)
+  end
+
   def command(command, data={})
     data["id"] ||= @multiplexer.generate_id
     data["command"] = command

vendor/distributor/lib/distributor/version.rb

@@ -1,5 +1,5 @@
 module Distributor
 
-  VERSION = "0.5.1"
+  VERSION = "0.6.0"
 
 end