REFACTOR: Use Async and family

Rakefile

@@ -2,6 +2,7 @@
 
 require "fileutils"
 require "rubygems"
+require 'rspec/core/rake_task'
 
 require "mail_catcher/version"
 

lib/mail_catcher.rb

@@ -1,26 +1,22 @@
 # frozen_string_literal: true
 
+require 'async/io/address_endpoint'
+require 'async/http/endpoint'
+require 'async/websocket/adapters/rack'
+require 'async/io/shared_endpoint'
+require 'falcon'
 require "open3"
 require "optparse"
 require "rbconfig"
-
-require "eventmachine"
-require "thin"
-
-module EventMachine
-  # Monkey patch fix for 10deb4
-  # See https://github.com/eventmachine/eventmachine/issues/569
-  def self.reactor_running?
-    (@reactor_running || false)
-  end
-end
+require 'socket'
+require 'mail'
 
 require "mail_catcher/version"
 
 module MailCatcher extend self
   autoload :Bus, "mail_catcher/bus"
   autoload :Mail, "mail_catcher/mail"
-  autoload :Smtp, "mail_catcher/smtp"
+  autoload :SMTP, "mail_catcher/smtp"
   autoload :Web, "mail_catcher/web"
 
   def env
@@ -173,75 +169,90 @@ def run! options=nil
 
     puts "Starting MailCatcher v#{VERSION}"
 
-    Thin::Logging.debug = development?
-    Thin::Logging.silent = !development?
+    Async.run do
+      @smtp_address = Async::IO::Address.tcp(options[:smtp_ip], options[:smtp_port])
+      @smtp_endpoint = Async::IO::AddressEndpoint.new(@smtp_address)
+      @smtp_socket = rescue_port(options[:smtp_port]) { @smtp_endpoint.bind }
+      puts "==> #{smtp_url}"
+
+      @http_address = Async::IO::Address.tcp(options[:http_ip], options[:http_port])
+      @http_endpoint = Async::IO::AddressEndpoint.new(@http_address)
+      @http_socket = rescue_port(options[:http_port]) { @http_endpoint.bind }
+      puts "==> #{http_url}"
+    end
 
-    # One EventMachine loop...
-    EventMachine.run do
-      # Set up an SMTP server to run within EventMachine
-      rescue_port options[:smtp_port] do
-        EventMachine.start_server options[:smtp_ip], options[:smtp_port], Smtp
-        puts "==> #{smtp_url}"
+    Async.logger.level = :debug if options[:verbose]
+
+    if options[:daemon]
+      if quittable?
+        puts "*** MailCatcher runs as a daemon by default. Go to the web interface to quit."
+      else
+        puts "*** MailCatcher is now running as a daemon that cannot be quit."
       end
+      Process.daemon
+    end
 
-      # Let Thin set itself up inside our EventMachine loop
-      # (Skinny/WebSockets just works on the inside)
-      rescue_port options[:http_port] do
-        Thin::Server.start(options[:http_ip], options[:http_port], Web)
-        puts "==> #{http_url}"
+    Async::Reactor.run do |task|
+      smtp_endpoint = MailCatcher::SMTP::URLEndpoint.new(URI.parse(smtp_url), @smtp_endpoint)
+      smtp_server = MailCatcher::SMTP::Server.new(smtp_endpoint) do |envelope|
+        MailCatcher::Mail.add_message(sender: envelope.sender, recipients: envelope.recipients,
+                                      source: envelope.content)
       end
 
-      # Open the web browser before detatching console
-      if options[:browse]
-        EventMachine.next_tick do
-          browse http_url
+      smtp_task = task.async do |task|
+        task.annotate "binding to #{@smtp_socket.local_address.inspect}"
+
+        begin
+          @smtp_socket.listen(Socket::SOMAXCONN)
+          @smtp_socket.accept_each(task: task, &smtp_server.method(:accept))
+        ensure
+          @smtp_socket.close
         end
       end
 
-      # Daemonize, if we should, but only after the servers have started.
-      if options[:daemon]
-        EventMachine.next_tick do
-          if quittable?
-            puts "*** MailCatcher runs as a daemon by default. Go to the web interface to quit."
-          else
-            puts "*** MailCatcher is now running as a daemon that cannot be quit."
-          end
-          Process.daemon
+      http_endpoint = Async::HTTP::Endpoint.new(URI.parse(http_url), @http_endpoint)
+      http_app = Falcon::Adapters::Rack.new(Web.app)
+      http_server = Falcon::Server.new(http_app, http_endpoint)
+
+      task.async do |task|
+        task.annotate "binding to #{@http_socket.local_address.inspect}"
+
+        begin
+          @http_socket.listen(Socket::SOMAXCONN)
+          @http_socket.accept_each(task: task, &http_server.method(:accept))
+        ensure
+          @http_socket.close
         end
       end
+
+      browse(http_url) if options[:browse]
     end
+  rescue Interrupt
+    # Cool story
   end
 
   def quit!
-    MailCatcher::Bus.push(type: "quit")
+    Async::Task.current.reactor.stop
+  end
 
-    EventMachine.next_tick { EventMachine.stop_event_loop }
+  def http_url
+    "http://#{@@options[:http_ip]}:#{@@options[:http_port]}#{@@options[:http_path]}"
   end
 
-protected
+  protected
 
   def smtp_url
     "smtp://#{@@options[:smtp_ip]}:#{@@options[:smtp_port]}"
   end
 
-  def http_url
-    "http://#{@@options[:http_ip]}:#{@@options[:http_port]}#{@@options[:http_path]}".chomp("/")
-  end
-
   def rescue_port port
     begin
       yield
-
-    # XXX: EventMachine only spits out RuntimeError with a string description
-    rescue RuntimeError
-      if $!.to_s =~ /\bno acceptor\b/
-        puts "~~> ERROR: Something's using port #{port}. Are you already running MailCatcher?"
-        puts "==> #{smtp_url}"
-        puts "==> #{http_url}"
-        exit -1
-      else
-        raise
-      end
+    rescue Errno::EADDRINUSE
+      puts "~~> ERROR: Something's using port #{port}. Are you already running MailCatcher?"
+      puts "==> #{smtp_url}"
+      puts "==> #{http_url}"
+      exit(-1)
     end
   end
 end

lib/mail_catcher/bus.rb

@@ -1,7 +1,48 @@
 # frozen_string_literal: true
 
-require "eventmachine"
-
 module MailCatcher
-  Bus = EventMachine::Channel.new
+  # Async-friendly broadcast channel
+  class Channel
+    def initialize
+      @subscription_id = 0
+      @subscriptions = {}
+    end
+
+    def subscriber_count
+      @subscriptions.size
+    end
+
+    def push(*values)
+      Async.run do
+        values.each do |value|
+          @subscriptions.each_value do |subscription|
+            Async do
+              subscription.call(value)
+            end
+          end
+        end
+      end
+    end
+
+    def subscribe(&block)
+      subscription_id = next_subscription_id
+
+      @subscriptions[subscription_id] = block
+
+      subscription_id
+    end
+
+    def unsubscribe(subscription_id)
+      @subscriptions.delete(subscription_id)
+    end
+
+    private
+
+    def next_subscription_id
+      @subscription_id += 1
+    end
+  end
+
+  # Then we instantiate a global one
+  Bus = Channel.new
 end

lib/mail_catcher/mail.rb

@@ -1,6 +1,5 @@
 # frozen_string_literal: true
 
-require "eventmachine"
 require "json"
 require "mail"
 require "sqlite3"
@@ -56,10 +55,7 @@ def add_message(message)
       add_message_part(message_id, cid, part.mime_type || "text/plain", part.attachment? ? 1 : 0, part.filename, part.charset, body, body.length)
     end
 
-    EventMachine.next_tick do
-      message = MailCatcher::Mail.message message_id
-      MailCatcher::Bus.push(type: "add", message: message)
-    end
+    MailCatcher::Bus.push(type: "add", message: message(message_id))
   end
 
   def add_message_part(*args)
@@ -157,18 +153,14 @@ def delete!
     @delete_all_messages_query ||= db.prepare "DELETE FROM message"
     @delete_all_messages_query.execute
 
-    EventMachine.next_tick do
-      MailCatcher::Bus.push(type: "clear")
-    end
+    MailCatcher::Bus.push(type: "clear")
   end
 
   def delete_message!(message_id)
     @delete_messages_query ||= db.prepare "DELETE FROM message WHERE id = ?"
     @delete_messages_query.execute(message_id)
 
-    EventMachine.next_tick do
-      MailCatcher::Bus.push(type: "remove", id: message_id)
-    end
+    MailCatcher::Bus.push(type: "remove", id: message_id)
   end
 
   def delete_older_messages!(count = MailCatcher.options[:messages_limit])