update & fix DevP2P::Peer sub protocols handling logic

lib/ethruby/devp2p/actor.rb

@@ -71,11 +71,14 @@ def queue
         end
       end
 
+      class Error < StandardError
+      end
+
       # stop actor
-      class StopError < StandardError
+      class StopError < Error
       end
 
-      class StateError < StandardError
+      class StateError < Error
       end
 
       attr_accessor :executor
@@ -110,6 +113,8 @@ def call(method, *args)
 
       # start actor
       def start
+        raise Error.new("must set executor before start") unless executor
+
         @running = true
         executor.post do
           start_loop
@@ -124,7 +129,7 @@ def start
       #   actor.wait
       #
       def send_stop
-        self << :raise_stop_error
+        self << [:raise_error, StopError.new]
       end
 
       # wait until an error occurs
@@ -137,7 +142,7 @@ def wait
       def start_loop
         loop_callback do
           # check inbox
-          next if @inbox.empty?
+          next Thread.pass if @inbox.empty?
           msg = @inbox.pop
 
           # extract sync or async call
@@ -183,9 +188,8 @@ def loop_callback
         yield
       end
 
-      private
-      def raise_stop_error
-        raise StopError.new
+      def raise_error(e)
+        raise e
       end
     end
 

lib/ethruby/devp2p/peer.rb

@@ -9,6 +9,8 @@ module DevP2P
     # represent a connected remote node
     class Peer
 
+      class DiscoverError < StandardError
+      end
       class UnknownMessageCodeError < StandardError
       end
 
@@ -54,17 +56,23 @@ def read_loop
           msg.received_at = Time.now
           handle(msg)
         end
+      rescue StandardError => e
+        self << [:raise_error, e]
       end
 
       def start_protocols
         @protocols.each do |protocol|
+          protocol.executor ||= executor
           protocol.start(self, @protocol_io_hash[protocol.name])
         end
       end
 
       def handle(msg)
         if msg.code == RLPX::MESSAGES[:ping]
           #TODO send pong
+        elsif msg.code == RLPX::MESSAGES[:discover]
+          reason = RLP.decode(msg.payload).ord
+          raise DiscoverError.new("receive error discovery message, reason: #{reason}")
         else
           # send msg to sub protocol
           if (protocol = find_protocol_by_msg_code(msg.code)).nil?
@@ -76,9 +84,11 @@ def handle(msg)
 
       private
       def find_protocol_by_msg_code(code)
-        @protocols.find do |protocol|
-          code >= protocol.offset && code <= protocol.offset + protocol.length
-        end
+        @protocol_io_hash.values.find do |protocol_io|
+          offset = protocol_io.offset
+          protocol = protocol_io.protocol
+          code >= offset && code < offset + protocol.length
+        end.protocol
       end
 
       # return protocol_io_hash
@@ -88,7 +98,7 @@ def make_protocol_io_hash(protocols, caps, io)
         offset = RLPX::BASE_PROTOCOL_LENGTH
         result = {}
         # [name, version] as key
-        protocols_hash = protocols.map {|protocol| [protocol.name, protocol.version]}.to_h
+        protocols_hash = protocols.map {|protocol| [[protocol.name, protocol.version], protocol]}.to_h
         sorted_caps = caps.sort_by {|c| [c.name, c.version]}
 
         sorted_caps.each do |cap|
@@ -97,7 +107,7 @@ def make_protocol_io_hash(protocols, caps, io)
           # ignore same name old protocols
           if (old = result[cap.name])
             result.delete(cap.name)
-            offset -= old.length
+            offset -= old.protocol.length
           end
           result[cap.name] = ProtocolIO.new(protocol, offset, io)
           # move offset, to support next protocol
@@ -108,4 +118,4 @@ def make_protocol_io_hash(protocols, caps, io)
     end
 
   end
-end
+end

lib/ethruby/devp2p/protocol.rb

@@ -1,20 +1,28 @@
 # frozen_string_literal: true
 
+require_relative 'actor'
+
 module ETH
   module DevP2P
 
     # protocol represent DevP2P sub protocols
     class Protocol
-      attr_accessor :name, :version, :length, :node_info, :peer_info
+
+      include Actor
+
+      attr_reader :name, :version, :length
+      attr_accessor :node_info, :peer_info
 
       def initialize(name:, version:, length:)
         @name = name
         @version = version
         @length = length
+        super()
       end
 
       # start protocol handling
-      def start
+      def start(peer, io)
+        super()
       end
     end
 

spec/ethruby/crypto_spec.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require 'spec_helper'
 require 'ethruby/crypto'
 require 'ethruby/key'
 

spec/ethruby/devp2p/actor_spec.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require 'spec_helper'
 require 'concurrent'
 require 'ethruby/devp2p/actor'
 

spec/ethruby/devp2p/peer_spec.rb

@@ -0,0 +1,102 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+require 'ethruby/devp2p/peer'
+require 'ethruby/devp2p/protocol'
+require 'ethruby/devp2p/rlpx/protocol_handshake'
+require 'concurrent'
+
+RSpec.describe ETH::DevP2P::Peer do
+  let(:executor) {Concurrent::CachedThreadPool.new}
+  let(:actor) {my_actor.new(executor: executor)}
+
+  after {executor.kill}
+
+  # mock connection
+  let(:connection) do
+    Class.new do
+      attr_reader :queue
+
+      def initialize
+        @queue = []
+      end
+
+      def read_msg
+        raise StandardError if queue.empty?
+        queue.shift
+      end
+    end.new
+  end
+
+  let(:mock_protocol) do
+    Class.new(ETH::DevP2P::Protocol) do
+      attr_reader :histories, :peer, :protocol_io
+
+      def initialize(*args)
+        super(*args)
+        @histories = []
+      end
+
+      def start(peer, io)
+        @peer = peer
+        @protocol_io = io
+        super
+      end
+
+      def handle_msg(msg)
+        histories << msg
+      end
+    end
+  end
+
+  it 'handle msg by code' do
+    protocol_1 = mock_protocol.new(name: 'eth', version: 63, length: 17)
+    protocol_2 = mock_protocol.new(name: 'eth', version: 62, length: 8)
+    protocol_3 = mock_protocol.new(name: 'hello', version: 1, length: 16)
+
+    caps = [
+      ETH::DevP2P::RLPX::Cap.new(name: 'eth', version: 63),
+      ETH::DevP2P::RLPX::Cap.new(name: 'eth', version: 62),
+      ETH::DevP2P::RLPX::Cap.new(name: 'hello', version: 1),
+    ]
+    handshake = ETH::DevP2P::RLPX::ProtocolHandshake.new(version: 4, name: 'test', caps: caps, id: 0)
+
+
+    msg_1 = ETH::DevP2P::RLPX::Message.new(code: 16, payload: "test_1".b, size: 6)
+    msg_2 = ETH::DevP2P::RLPX::Message.new(code: 32, payload: "test_2".b, size: 6)
+    msg_3 = ETH::DevP2P::RLPX::Message.new(code: 33, payload: "test_hello".b, size: 10)
+
+    # send messages to connection
+    connection.queue << msg_1
+    connection.queue << msg_2
+    connection.queue << msg_3
+
+    peer = ETH::DevP2P::Peer.new(connection, handshake, [protocol_1, protocol_2, protocol_3])
+    peer.executor = executor
+    peer.start
+
+    # peer read all messages
+    expect {peer.wait}.to raise_error(StandardError)
+
+    # 'eth' protocol
+    protocol_1.send_stop
+    protocol_1.wait
+    expect(protocol_1.peer).to be peer
+    expect(protocol_1.protocol_io.io).to be connection
+    expect(protocol_1.histories).to eq [msg_1, msg_2]
+
+    # old 'eth' protocol
+    protocol_2.send_stop
+    protocol_2.wait
+    expect(protocol_2.peer).to be peer
+    expect(protocol_2.protocol_io.io).to be connection
+    expect(protocol_2.histories).to eq []
+
+    # 'hello' protocol
+    protocol_3.send_stop
+    protocol_3.wait
+    expect(protocol_3.peer).to be peer
+    expect(protocol_3.protocol_io.io).to be connection
+    expect(protocol_3.histories).to eq [msg_3]
+  end
+end

spec/ethruby_spec.rb

@@ -1,3 +1,5 @@
+require 'spec_helper'
+
 RSpec.describe ETH do
   it "has a version number" do
     expect(ETH::VERSION).not_to be nil

spec/spec_helper.rb

@@ -11,4 +11,8 @@
   config.expect_with :rspec do |c|
     c.syntax = :expect
   end
+
+  # set concurrent ruby logger
+  require 'concurrent'
+  Concurrent.use_simple_logger(level = Logger::DEBUG)
 end