Implements concurrent requests

src/scry.cr

@@ -4,48 +4,13 @@ require "./scry/context"
 require "./scry/message"
 require "./scry/response"
 require "./scry/environment_config"
+require "./scry/concurrent_rpc"
 
 module Scry
   def self.start
     Log.logger.info("Scry is looking into your code...")
-
-    at_exit do
-      Log.logger.info("...your session has ended")
-    end
-
     EnvironmentConfig.new.run
-
-    next_request = Channel(Int32).new(10)
-    context = Context.new
-
-    loop do |i|
-      spawn do
-        Log.logger.debug("Scry is listening request ##{i}...")
-        content = Request.new(STDIN).read
-        request = Message.new(content).parse
-        results = context.dispatch(request)
-      rescue ex
-        results = [ResponseMessage.new(ex)]
-      ensure
-        response = Response.new([results].flatten)
-        response.write(STDOUT)
-        next_request.send i
-      end
-
-      select
-      when n = next_request.receive
-        Log.logger.debug("Scry has processed request ##{n}!")
-        next
-      else
-        sleep 1
-      end
-    end
-  rescue ex
-    Log.logger.error(
-      %(#{ex.message || "Unknown error"}\n#{ex.backtrace.join('\n')})
-    ) unless Log.logger.nil?
-  ensure
-    Log.logger.close unless Log.logger.nil?
+    ConcurrentRpc.new.run
   end
 end
 

src/scry/concurrent_rpc.cr

@@ -0,0 +1,69 @@
+module Scry
+  class ConcurrentRpc
+    CONTEXT = Context.new
+
+    private def get_requests(reader, input_channel)
+      Log.logger.info "getting request..."
+      content = Request.new(reader).read
+      Log.logger.info "getting content..."
+      input_channel.send content
+    end
+
+    private def handle_request(content, result_channel)
+      Log.logger.info "Processing request..."
+      Log.logger.info content
+      request = Message.new(content).parse
+      results = CONTEXT.dispatch(request)
+    rescue ex
+      results = [ResponseMessage.new(ex)]
+    ensure
+      Log.logger.info results
+      Log.logger.info "End Processing request..."
+      result_channel.send [results].flatten
+    end
+
+    def run
+      Log.logger.info "server has started"
+      input_channel = Channel(String | Nil).new
+
+      spawn get_requests(reader, input_channel)
+
+      channels = [] of Channel(String | Nil) | Channel(Array(Result))
+      channels << input_channel
+
+      until channels.empty?
+        channel_index, data = Channel.select(channels.map &.receive_select_action)
+        Log.logger.info "data received"
+        Log.logger.info channel_index
+        Log.logger.info data
+
+        if channels[channel_index] == input_channel
+          if data.nil?
+            Log.logger.info "Input closed, no more data to come in current channel..."
+            channels.delete_at(channel_index)
+            next
+          end
+
+          content = data.as(String)
+          result_channel = Channel(Array(Result)).new
+
+          spawn handle_request(content, result_channel)
+
+          channels << result_channel
+        else
+          results = data.as(Array(Result))
+          Log.logger.info "Printing result output..."
+          Log.logger.info results
+          response = Response.new(results)
+          response.write(writer)
+          channels.delete_at(channel_index)
+        end
+      end
+    rescue ex
+      Log.logger.error(ex.inspect_with_backtrace) unless Log.logger.nil?
+    ensure
+      Log.logger.info("...your session has ended")
+      Log.logger.close unless Log.logger.nil?
+    end
+  end
+end