Extract concurrency logic to a separated class.

Remove unnecesary additional vendorized files.

Use Celluloid in Ruby 1.9.

engineyard-serverside.gemspec

@@ -16,8 +16,9 @@ Gem::Specification.new do |s|
   s.executables = ["engineyard-serverside"]
   s.default_executable = "engineyard-serverside"
   s.require_path = 'lib'
-  s.add_development_dependency('rspec', '= 1.3.2')
-  s.add_development_dependency('rake', '>= 0.9.2.2')
+
+  s.add_development_dependency('rspec', '1.3.2')
+  s.add_development_dependency('rake', '>=0.9.2.2')
 
   s.required_rubygems_version = %q{>= 1.3.6}
   s.test_files = Dir.glob("spec/**/*")

lib/engineyard-serverside.rb

@@ -1,21 +1,15 @@
-if String.instance_methods.include?(:force_encoding)
-  $string_encodings = true
-else
-  # KCODE is gone in 1.9-like implementations, but we
-  # still need to set it for 1.8.
-  $KCODE = 'U'
-  $string_encodings = false
-end
+# encoding: utf-8
 
-$LOAD_PATH.unshift File.expand_path('vendor/thor/lib', File.dirname(__FILE__))
-$LOAD_PATH.unshift File.expand_path('vendor/open4/lib', File.dirname(__FILE__))
 $LOAD_PATH.unshift File.expand_path('vendor/escape/lib', File.dirname(__FILE__))
 $LOAD_PATH.unshift File.expand_path('vendor/json_pure/lib', File.dirname(__FILE__))
-$LOAD_PATH.unshift File.expand_path('vendor/dataflow', File.dirname(__FILE__))
+$LOAD_PATH.unshift File.expand_path('vendor/open4/lib', File.dirname(__FILE__))
+$LOAD_PATH.unshift File.expand_path('vendor/thor/lib', File.dirname(__FILE__))
 
 require 'escape'
 require 'json'
-require 'dataflow'
+require 'tmpdir'
+require 'thor'
+require 'open4'
 
 require 'engineyard-serverside/version'
 require 'engineyard-serverside/strategies/git'
@@ -27,6 +21,7 @@
 require 'engineyard-serverside/cli'
 require 'engineyard-serverside/configuration'
 require 'engineyard-serverside/deprecation'
+require 'engineyard-serverside/future'
 
 module EY
   module Serverside
@@ -60,13 +55,16 @@ def self.dna_path
     end
 
     def self.read_encoded_dna
-      json = if File.exist?(dna_path)
-               `sudo cat #{dna_path}`
-             else
-               '{}'
-             end
-      json.force_encoding('UTF-8') if $string_encodings
-      json
+      encoded_dna = '{}'
+      force_unicode = encoded_dna.respond_to?(:force_encoding)
+      $KCODE = 'U' unless force_unicode
+
+      if File.exist?(dna_path)
+        encoded_dna = `sudo cat #{dna_path}`
+        encoded_dna.force_encoding('UTF-8') if force_unicode
+      end
+
+      encoded_dna
     end
   end
 end

lib/engineyard-serverside/cli.rb

@@ -1,16 +1,8 @@
-require 'thor'
 require 'pathname'
 
 module EY
   module Serverside
     class CLI < Thor
-      include Dataflow
-
-      def self.start(*)
-        super
-      rescue RemoteFailure
-        exit(1)
-      end
 
       method_option :migrate,         :type     => :string,
                                       :desc     => "Run migrations with this deploy",
@@ -56,14 +48,14 @@ def self.start(*)
                                       :aliases  => ["-v"]
 
       desc "deploy", "Deploy code from /data/<app>"
-      def deploy(default_task=:deploy)
+      def deploy(default_task = :deploy)
         config = EY::Serverside::Deploy::Configuration.new(options)
         EY::Serverside::Server.load_all_from_array(assemble_instance_hashes(config))
 
         EY::Serverside::LoggedOutput.verbose = options[:verbose]
         EY::Serverside::LoggedOutput.logfile = File.join(ENV['HOME'], "#{options[:app]}-deploy.log")
 
-        invoke :propagate
+        propagate
 
         EY::Serverside::Deploy.new(config).send(default_task)
       end
@@ -144,7 +136,7 @@ def integrate
 
         EY::Serverside::Server.load_all_from_array(assemble_instance_hashes(config))
 
-        invoke :propagate
+        propagate
 
         EY::Serverside::Server.all.each do |server|
           server.sync_directory app_dir
@@ -190,7 +182,7 @@ def restart
         config = EY::Serverside::Deploy::Configuration.new(options)
         EY::Serverside::Server.load_all_from_array(assemble_instance_hashes(config))
 
-        invoke :propagate
+        propagate
 
         EY::Serverside::Deploy.new(config).restart_with_maintenance_page
       end
@@ -212,34 +204,22 @@ def install_bundler(version)
 
       desc "propagate", "Propagate the engineyard-serverside gem to the other instances in the cluster. This will install exactly version #{EY::Serverside::VERSION}."
       def propagate
-        config          = EY::Serverside::Deploy::Configuration.new
-        gem_filename    = "engineyard-serverside-#{EY::Serverside::VERSION}.gem"
-        local_gem_file  = File.join(Gem.dir, 'cache', gem_filename)
-        remote_gem_file = File.join(Dir.tmpdir, gem_filename)
-        gem_binary      = File.join(Gem.default_bindir, 'gem')
-
-        barrier(*(EY::Serverside::Server.all.find_all do |server|
-          !server.local?            # of course this machine has it
-        end.map do |server|
-          need_later do
-            egrep_escaped_version = EY::Serverside::VERSION.gsub(/\./, '\.')
-            # the [,)] is to stop us from looking for e.g. 0.5.1, seeing
-            # 0.5.11, and mistakenly thinking 0.5.1 is there
-            has_gem_cmd = "#{gem_binary} list engineyard-serverside | grep \"engineyard-serverside\" | egrep -q '#{egrep_escaped_version}[,)]'"
-
-            if !server.run(has_gem_cmd)  # doesn't have this exact version
-              puts "~> Installing engineyard-serverside on #{server.hostname}"
-
-              system(Escape.shell_command([
-                'scp', '-i', "#{ENV['HOME']}/.ssh/internal",
-                "-o", "StrictHostKeyChecking=no",
-                local_gem_file,
-               "#{config.user}@#{server.hostname}:#{remote_gem_file}",
-              ]))
-              server.run("sudo #{gem_binary} install --no-rdoc --no-ri '#{remote_gem_file}'")
+        ey_server_side = Dependency.new('engineyard-serverside', EY::Serverside::VERSION)
+        servers = EY::Serverside::Server.all.find_all { |server| !server.local? }
+
+        futures = EY::Serverside::Future.call(servers) do |server|
+          installed = server.gem?(ey_server_side.name, ey_server_side.version)
+          unless installed
+            unless server.gem?(ey_server_side.name, ey_server_side.version)
+              puts "~> Installing #{ey_server_side.name} on #{server.hostname}"
+              server.copy(ey_server_side.local_path, ey_server_side.remote_path)
+              installed = server.install_gem(ey_server_side.remote_path)
             end
           end
-        end))
+          installed
+        end
+
+        EY::Serverside::Future.success?(futures)
       end
 
       private
@@ -253,6 +233,18 @@ def assemble_instance_hashes(config)
           }
         }
       end
+
+      class Dependency
+        attr_reader :name, :version
+        def initialize(name, version)
+          @name = name
+          @version = version
+        end
+
+        def gemname; "#{name}-#{version}.gem"; end
+        def local_path; File.expand_path(File.join('cache', gemname), Gem.dir); end
+        def remote_path; File.expand_path(gemname, Dir.tmpdir); end
+      end
     end
   end
 end

lib/engineyard-serverside/configuration.rb

@@ -1,13 +1,37 @@
-require 'json'
-require 'thor'
 
 module EY
   module Serverside
+    # Really simple and hacked implementation that works for the 99.9999% of the cases
+    class HashWithIndifferentAccess
+      def initialize(hash = {})
+        @internal = {}
+        hash.each do |k, v|
+          @internal[k.to_s] = v
+        end
+      end
+      def [](name)
+        @internal[name.to_s]
+      end
+      def []=(name, value)
+        @internal[name.to_s] = value
+      end
+
+      def method_missing(method, *args, &block)
+        if @internal.key?(method.to_s)
+          @internal[method.to_s]
+        else
+          @internal.send(method, args, &block)
+        end
+      end
+    end
+
     class Deploy::Configuration
+      require 'json'
+
       DEFAULT_CONFIG = Thor::CoreExt::HashWithIndifferentAccess.new({
         "branch"         => "master",
         "strategy"       => "Git",
-        "bundle_without" => "test development",
+        "bundle_without" => "test development"
       })
 
       attr_reader :configuration

lib/engineyard-serverside/deploy.rb

@@ -145,9 +145,10 @@ def run_with_callbacks(task)
       # task
       def push_code
         info "~> Pushing code to all servers"
-        barrier *(EY::Serverside::Server.all.map do |server|
-          need_later { server.sync_directory(config.repository_cache) }
-        end)
+        futures = EY::Serverside::Future.call(EY::Serverside::Server.all) do |server|
+          server.sync_directory(config.repository_cache)
+        end
+        EY::Serverside::Future.success?(futures)
       end
 
       # task

lib/engineyard-serverside/future.rb

@@ -0,0 +1,29 @@
+module EY
+  module Serverside
+    class Future
+      def self.success?(futures)
+        futures.empty? || futures.all? {|f| f.success?}
+      end
+
+      def initialize(server, *args, &block)
+        @server = server
+        @args = args
+        @block = block
+      end
+
+      def success?
+        @value == true
+      end
+
+      def error?
+        !success?
+      end
+    end
+
+    if defined?(Fiber)
+      require 'engineyard-serverside/futures/celluloid'
+    else
+      require 'engineyard-serverside/futures/dataflow'
+    end
+  end
+end

lib/engineyard-serverside/futures/celluloid.rb

@@ -0,0 +1,25 @@
+module EY
+  module Serverside
+    $LOAD_PATH.unshift File.expand_path('../../vendor/celluloid/lib', File.dirname(__FILE__))
+    require 'celluloid'
+    class Future
+      def self.call(servers, *args, &block)
+        futures = servers.map do |server|
+          new(server, *args, &block)
+        end
+
+        futures.each {|f| f.call}
+        futures
+      end
+
+      def future
+        Celluloid::Future.new(@server, *@args, &@block)
+      end
+
+      def call
+        # Celluloid needs to call the block explicitely
+        @value ||= future.call
+      end
+    end
+  end
+end

lib/engineyard-serverside/futures/dataflow.rb

@@ -0,0 +1,31 @@
+module EY
+  module Serverside
+    $LOAD_PATH.unshift File.expand_path('../../vendor/dataflow', File.dirname(__FILE__))
+    require 'dataflow'
+
+    class Future
+      extend Dataflow
+
+      def self.call(servers, *args, &block)
+        futures = []
+        # Dataflow needs to call `barrier` and `need_later` in the same object
+        barrier(*servers.map do |server|
+          future = new(server, *args, &block)
+          futures << future
+
+          need_later { future.call }
+        end)
+
+        futures
+      end
+
+      def future
+        @block.call(@server, *@args)
+      end
+
+      def call
+        @value ||= future
+      end
+    end
+  end
+end

lib/engineyard-serverside/logged_output.rb

@@ -1,5 +1,3 @@
-require 'open4'
-
 module EY
   module Serverside
     module LoggedOutput

lib/engineyard-serverside/server.rb

@@ -83,12 +83,36 @@ def run(command)
         if local?
           logged_system(command)
         else
-          logged_system(ssh_command + " " + Escape.shell_command(["#{user}@#{hostname}", command]))
+          logged_system(ssh_command + Escape.shell_command(["#{user}@#{hostname}", command]))
         end
       end
 
+      def copy(local_file, remote_file)
+        logged_system(scp_command + Escape.shell_command([local_file, "#{user}@#{hostname}:#{remote_file}"]))
+      end
+
       def ssh_command
-        "ssh -i #{ENV['HOME']}/.ssh/internal -o StrictHostKeyChecking=no -o PasswordAuthentication=no"
+        "ssh #{ssh_options} "
+      end
+
+      def scp_command
+        "scp #{ssh_options} "
+      end
+
+      def ssh_options
+        "-i #{ENV['HOME']}/.ssh/internal -o StrictHostKeyChecking=no -o PasswordAuthentication=no"
+      end
+
+      def gem?(name, version)
+        run("#{gem_command} list -i #{name} -v '#{version}'")
+      end
+
+      def install_gem(path)
+        run("#{gem_command} install -q --no-ri --no-rdoc #{path}")
+      end
+
+      def gem_command
+        File.expand_path('gem', Gem.default_bindir)
       end
 
     end