Server/Client: add each_key support

feature_matrix.yaml

@@ -219,13 +219,14 @@ backends:
   - adapter: Client
     platforms: [ MRI, JRuby ]
     features: [multiprocess]
-    unknown: [ increment, create, expires, persist ]
+    unknown: [ increment, create, expires, persist, each_key ]
     description: "Moneta client adapter"
     notes:
       increment: depends on server
       create: depends on server
       expires: depends on server
       persist: depends on server
+      each_key: depends on server
   - adapter: RestClient
     platforms: [ MRI, JRuby ]
     features: [ multiprocess ]

lib/moneta/adapters/client.rb

@@ -70,6 +70,37 @@ def close
         nil
       end
 
+      # (see Proxy#each_key)
+      def each_key
+        raise NotImplementedError, 'each_key is not supported' unless supports?(:each_key)
+        return enum_for(:each_key) unless block_given?
+
+        begin
+          write(:each_key)
+          yield_break = false
+
+          loop do
+            write('NEXT')
+
+            # A StopIteration error will be raised by this call if the server
+            # reached the end of the enumeration.  This will stop the loop
+            # automatically.
+            result = read_msg
+
+            # yield_break will be true in the ensure block (below) if anything
+            # happened during the yield to stop further enumeration.
+            yield_break = true
+            yield result
+            yield_break = false
+          end
+        ensure
+          write('BREAK') if yield_break
+          read_msg # nil return from each_key
+        end
+
+        self
+      end
+
       # (see Default#features)
       def features
         @features ||=

lib/moneta/lock.rb

@@ -15,15 +15,20 @@ def initialize(adapter, options = {})
     protected
 
     def wrap(name, *args, &block)
+      self.locks ||= Set.new
       if locked?
         yield
       else
         lock!(&block)
       end
     end
 
+    def locks=(locks)
+      Thread.current.thread_variable_set('Moneta::Lock', locks)
+    end
+
     def locks
-      Thread.current['Moneta::Lock'] ||= Set.new
+      Thread.current.thread_variable_get('Moneta::Lock')
     end
 
     def lock!(&block)

lib/moneta/server.rb

@@ -30,14 +30,14 @@ def mainloop
 
       def dispatch(method, args)
         case method
-        when :key?, :load, :delete, :increment, :create
+        when :key?, :load, :delete, :increment, :create, :features
           @store.send(method, *args)
-        when :features
-          # all features except each_key are supported
-          @store.features - [:each_key]
         when :store, :clear
           @store.send(method, *args)
           nil
+        when :each_key
+          yield_each(@store.each_key)
+          nil
         end
       rescue => e
         e
@@ -102,6 +102,29 @@ def pack(obj)
         s = Marshal.dump(obj)
         [s.bytesize].pack('N') << s
       end
+
+      def yield_each(enumerator)
+        received_break = false
+        loop do
+          case msg = read_msg
+          when %w{NEXT}
+            # This will raise a StopIteration at the end of the enumeration,
+            # which will exit the loop.
+            write(enumerator.next)
+          when %w{BREAK}
+            # This is received when the client wants to stop the enumeration.
+            received_break = true
+            break
+          else
+            # Otherwise, the client is attempting to call another method within
+            # an `each` block.
+            write_dispatch(msg)
+          end
+        end
+      ensure
+        # This tells the client to stop enumerating
+        write(StopIteration.new("Server initiated stop")) unless received_break
+      end
     end
 
     # @param [Hash] options

lib/moneta/shared.rb

@@ -11,8 +11,6 @@ module Moneta
   #
   # @api public
   class Shared < Wrapper
-    not_supports :each_key
-
     # @param [Hash] options
     # @option options [Integer] :port (9000) TCP port
     # @option options [String] :host Server hostname

spec/helper.rb

@@ -327,8 +327,13 @@ def start_server(*args)
       server = Moneta::Server.new(*args)
       Thread.new { server.run }
       sleep 0.1 until server.running?
+      server
     rescue Exception => ex
       puts "Failed to start server - #{ex.message}"
+      tries ||= 0
+      tries += 1
+      sleep Moneta::Server::TIMEOUT
+      tries < 3 ? retry : raise
     end
 
     def moneta_property_of(keys: 0, values: 0)

spec/moneta/adapters/client/adapter_client_spec.rb

@@ -1,11 +1,15 @@
 describe 'adapter_client', isolate: true, adapter: :Client do
   before :all do
-    start_server(Moneta::Adapters::Memory.new)
+    @server = start_server(Moneta::Adapters::Memory.new)
+  end
+
+  after :all do
+    @server.stop
   end
 
   moneta_build do
     Moneta::Adapters::Client.new
   end
 
-  moneta_specs ADAPTER_SPECS
+  moneta_specs ADAPTER_SPECS.with_each_key
 end

spec/moneta/adapters/client/standard_client_tcp_spec.rb

@@ -1,6 +1,10 @@
 describe "standard_client_tcp", isolate: true, adapter: :Client do
   before :all do
-    start_server(Moneta::Adapters::Memory.new)
+    @server = start_server(Moneta::Adapters::Memory.new)
+  end
+
+  after :all do
+    @server.stop
   end
 
   moneta_store :Client

spec/moneta/adapters/client/standard_client_unix_spec.rb

@@ -1,7 +1,11 @@
 describe "standard_client_unix", isolate: true, adapter: :Client do
   before :all do
-    start_server Moneta::Adapters::Memory.new,
-                 socket: File.join(tempdir, 'standard_client_unix')
+    @server = start_server Moneta::Adapters::Memory.new,
+                           socket: File.join(tempdir, 'standard_client_unix')
+  end
+
+  after :all do
+    @server.stop
   end
 
   moneta_store :Client do

spec/moneta/proxies/shared/shared_tcp_spec.rb

@@ -3,13 +3,13 @@
     tempdir = self.tempdir
     Moneta.build do
       use(:Shared, port: 9001) do
-        adapter :PStore, file: File.join(tempdir, 'shared_tcp')
+        adapter :GDBM, file: File.join(tempdir, 'shared_tcp')
       end
     end
   end
 
   shared_examples :shared_tcp do
-    moneta_specs ADAPTER_SPECS
+    moneta_specs ADAPTER_SPECS.with_each_key
 
     it 'shares values' do
       store['shared_key'] = 'shared_value'