#748

Committed 24 Apr 2014 12:31AM UTC coverage: 78.31% (-19.5%) from 97.805%

Build # #748

Build Type

push

Committed by

Commit Message

Attempting to fix a brittle test of Concurrent::timer.

Coverage Stats

1928 of 2462 relevant lines covered (78.31%)

409.47 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

31.58

/lib/concurrent/actor/simple_actor_ref.rb

require 'concurrent/actor/actor_ref'
require 'concurrent/atomic/event'
require 'concurrent/executor/single_thread_executor'
require 'concurrent/ivar'

module Concurrent

  class SimpleActorRef
    include ActorRef

    def initialize(actor, opts = {})
      @actor = actor
      @mutex = Mutex.new
      @executor = SingleThreadExecutor.new
      @stop_event = Event.new
      @reset_on_error = opts.fetch(:reset_on_error, true)
      @exception_class = opts.fetch(:rescue_exception, false) ? Exception : StandardError
      @args = opts.fetch(:args, []) if @reset_on_error

      @actor.define_singleton_method(:shutdown, &method(:set_stop_event))
      @actor.on_start
    end

    def running?
      ! @stop_event.set?
    end

    def shutdown?
      @stop_event.set?
    end

    def post(*msg, &block)
      raise ArgumentError.new('message cannot be empty') if msg.empty?
      ivar = IVar.new
      @executor.post(Message.new(msg, ivar, block), &method(:process_message))
      ivar
    end

    def post!(timeout, *msg)
      raise Concurrent::TimeoutError unless timeout.nil? || timeout >= 0
      ivar = self.post(*msg)
      ivar.value(timeout)
      if ivar.incomplete?
        raise Concurrent::TimeoutError
      elsif ivar.reason
        raise ivar.reason
      end
      ivar.value
    end

    def shutdown
      @mutex.synchronize do
        return if shutdown?
        @executor.shutdown 
        @actor.on_shutdown
        @stop_event.set
      end
    end

    def join(limit = nil)
      @stop_event.wait(limit)
    end

    private

    Message = Struct.new(:payload, :ivar, :callback)

    def set_stop_event
      @stop_event.set
    end

    def process_message(message)
      result = ex = nil

      begin
        result = @actor.receive(*message.payload)
      rescue @exception_class => ex
        @actor.on_error(Time.now, message.payload, ex)
        if @reset_on_error
          @mutex.synchronize{ @actor = @actor.class.new(*@args) }
        end
      ensure
        now = Time.now
        message.ivar.complete(ex.nil?, result, ex)

        begin
          message.callback.call(now, result, ex) if message.callback
        rescue @exception_class => ex
          # suppress
        end
      end
    end
  end
end

1	require 'concurrent/actor/actor_ref'	1✔
2	require 'concurrent/atomic/event'	1✔
3	require 'concurrent/executor/single_thread_executor'	1✔
4	require 'concurrent/ivar'	1✔
5
6	module Concurrent	1✔
7
8	class SimpleActorRef	1✔
9	include ActorRef	1✔
10
11	def initialize(actor, opts = {})	1✔
12	@actor = actor	×
13	@mutex = Mutex.new	×
14	@executor = SingleThreadExecutor.new	×
15	@stop_event = Event.new	×
16	@reset_on_error = opts.fetch(:reset_on_error, true)	×
17	@exception_class = opts.fetch(:rescue_exception, false) ? Exception : StandardError	×
18	@args = opts.fetch(:args, []) if @reset_on_error	×
19
20	@actor.define_singleton_method(:shutdown, &method(:set_stop_event))	×
21	@actor.on_start	×
22	end
23
24	def running?	1✔
25	! @stop_event.set?	×
26	end
27
28	def shutdown?	1✔
29	@stop_event.set?	×
30	end
31
32	def post(*msg, &block)	1✔
33	raise ArgumentError.new('message cannot be empty') if msg.empty?	×
34	ivar = IVar.new	×
35	@executor.post(Message.new(msg, ivar, block), &method(:process_message))	×
36	ivar	×
37	end
38
39	def post!(timeout, *msg)	1✔
40	raise Concurrent::TimeoutError unless timeout.nil? \|\| timeout >= 0	×
41	ivar = self.post(*msg)	×
42	ivar.value(timeout)	×
43	if ivar.incomplete?	×
44	raise Concurrent::TimeoutError	×
45	elsif ivar.reason	×
46	raise ivar.reason	×
47	end
48	ivar.value	×
49	end
50
51	def shutdown	1✔
52	@mutex.synchronize do	×
53	return if shutdown?	×
54	@executor.shutdown	×
55	@actor.on_shutdown	×
56	@stop_event.set	×
57	end
58	end
59
60	def join(limit = nil)	1✔
61	@stop_event.wait(limit)	×
62	end
63
64	private	1✔
65
66	Message = Struct.new(:payload, :ivar, :callback)	1✔
67
68	def set_stop_event	1✔
69	@stop_event.set	×
70	end
71
72	def process_message(message)	1✔
73	result = ex = nil	×
74
75	begin
76	result = @actor.receive(*message.payload)	×
77	rescue @exception_class => ex	×
78	@actor.on_error(Time.now, message.payload, ex)	×
79	if @reset_on_error	×
80	@mutex.synchronize{ @actor = @actor.class.new(*@args) }	×
81	end
82	ensure
83	now = Time.now	×
84	message.ivar.complete(ex.nil?, result, ex)	×
85
86	begin
87	message.callback.call(now, result, ex) if message.callback	×
88	rescue @exception_class => ex
89	# suppress
90	end
91	end
92	end
93	end
94	end

jdantonio / concurrent-ruby / #748

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous