guard / listen / 800

pending completion

Build # 800

Build Type

Pull #242

travis-ci

Committed by

francois2metz

Commit Message

<a href="https://github.com/guard/listen/commit/<a class=hub.com/guard/listen/commit/2b4a928a44050ac69685d7fd51aa4fea3732e771">2b4a928a4<a href="https://github.com/guard/listen/commit/2b4a928a44050ac69685d7fd51aa4fea3732e771">&quot;&gt;Revert &amp;quot;Use options of the adapter.&quot;.

This reverts commit </a><a class="double-link" href="https://github.com/guard/listen/commit/<a class="double-link" href="https://github.com/guard/listen/commit/653655f598afebee24a8469bfed24f570ead44df">653655f59</a>">653655f59</a><a href="https://github.com/guard/listen/commit/2b4a928a44050ac69685d7fd51aa4fea3732e771">.

Pull Request Pull Request #242: Fix tcp receiver host and port.

Run Details

11 of 11 new or added lines in 3 files covered. (100.0%)

2187 of 2373 relevant lines covered (92.16%)

84.83 hits per line

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

77.56

/lib/listen/listener.rb

require 'pathname'
require 'listen/adapter'
require 'listen/change'
require 'listen/record'
require 'listen/silencer'
require 'listen/queue_optimizer'
require 'English'

module Listen
  class Listener
    include Celluloid::FSM
    include QueueOptimizer

    attr_accessor :block

    attr_reader :silencer

    # TODO: deprecate
    attr_reader :options, :directories
    attr_reader :registry, :supervisor

    # Initializes the directories listener.
    #
    # @param [String] directory the directories to listen to
    # @param [Hash] options the listen options (see Listen::Listener::Options)
    #
    # @yield [modified, added, removed] the changed files
    # @yieldparam [Array<String>] modified the list of modified files
    # @yieldparam [Array<String>] added the list of added files
    # @yieldparam [Array<String>] removed the list of removed files
    #
    def initialize(*args, &block)
      @options = _init_options(args.last.is_a?(Hash) ? args.pop : {})

      # Setup logging first
      Celluloid.logger.level = _debug_level
      _log :info, "Celluloid loglevel set to: #{Celluloid.logger.level}"

      @silencer = Silencer.new
      _reconfigure_silencer({})

      @tcp_mode = nil
      if [:recipient, :broadcaster].include? args[1]
        target = args.shift
        @tcp_mode = args.shift
        _init_tcp_options(target)
      end

      @directories = args.flatten.map { |path| Pathname.new(path).realpath }
      @queue = Queue.new
      @block = block
      @registry = Celluloid::Registry.new

      transition :stopped
    end

    default_state :initializing

    state :initializing, to: :stopped
    state :paused, to: [:processing, :stopped]

    state :stopped, to: [:processing] do
      _stop_wait_thread
      if @supervisor
        @supervisor.terminate
        @supervisor = nil
      end
    end

    state :processing, to: [:paused, :stopped] do
      if wait_thread # means - was paused
        _wakeup_wait_thread
      else
        @last_queue_event_time = nil
        _start_wait_thread
        _init_actors

        # Note: make sure building is finished before starting adapter (for
        # consistent results both in specs and normal usage)
        sync(:record).build

        _start_adapter
      end
    end

    # Starts processing events and starts adapters
    # or resumes invoking callbacks if paused
    def start
      transition :processing
    end

    # TODO: depreciate
    alias_method :unpause, :start

    # Stops processing and terminates all actors
    def stop
      transition :stopped
    end

    # Stops invoking callbacks (messages pile up)
    def pause
      transition :paused
    end

    # processing means callbacks are called
    def processing?
      state == :processing
    end

    def paused?
      state == :paused
    end

    # TODO: deprecate
    alias_method :listen?, :processing?

    # TODO: deprecate
    def paused=(value)
      transition value ? :paused : :processing
    end

    # TODO: deprecate
    alias_method :paused, :paused?

    # Add files and dirs to ignore on top of defaults
    #
    # (@see Listen::Silencer for default ignored files and dirs)
    #
    def ignore(regexps)
      _reconfigure_silencer(ignore: [options[:ignore], regexps])
    end

    # Replace default ignore patterns with provided regexp
    def ignore!(regexps)
      _reconfigure_silencer(ignore: [], ignore!: regexps)
    end

    # Listen only to files and dirs matching regexp
    def only(regexps)
      _reconfigure_silencer(only: regexps)
    end

    def async(type)
      proxy = sync(type)
      proxy ? proxy.async : nil
    end

    def sync(type)
      @registry[type]
    end

    def queue(type, change, dir, path, options = {})
      fail "Invalid type: #{type.inspect}" unless [:dir, :file].include? type
      fail "Invalid change: #{change.inspect}" unless change.is_a?(Symbol)
      fail "Invalid path: #{path.inspect}" unless path.is_a?(String)
      @queue << [type, change, dir, path, options]

      @last_queue_event_time = Time.now.to_f
      _wakeup_wait_thread unless state == :paused

      return unless @tcp_mode == :broadcaster

      message = TCP::Message.new(type, change, dir, path, options)
      registry[:broadcaster].async.broadcast(message.payload)
    end

    private

    def _init_options(options = {})
      { debug: false,
        latency: nil,
        wait_for_delay: 0.1,
        force_polling: false,
        polling_fallback_message: nil }.merge(options)
    end

    def _debug_level
      debugging = ENV['LISTEN_GEM_DEBUGGING'] || options[:debug]
      case debugging.to_s
      when /2/
        Logger::DEBUG
      when /true|yes|1/i
        Logger::INFO
      else
        Logger::ERROR
      end
    end

    def _init_actors
      @supervisor = Celluloid::SupervisionGroup.run!(registry)
      supervisor.add(Record, as: :record, args: self)
      supervisor.pool(Change, as: :change_pool, args: self)

      if @tcp_mode == :broadcaster
        require 'listen/tcp/broadcaster'
        supervisor.add(TCP::Broadcaster, as: :broadcaster, args: [@options[:host], @options[:port]])

        # TODO: should be auto started, because if it crashes
        # a new instance is spawned by supervisor, but it's 'start' isn't
        # called
        registry[:broadcaster].start
      end

      options = [mq: self, directories: directories]
      supervisor.add(_adapter_class, as: :adapter, args: options)
    end

    def _wait_for_changes
      latency = options[:wait_for_delay]

      loop do
        break if state == :stopped

        if state == :paused || @queue.empty?
          sleep
          break if state == :stopped
        end

        # Assure there's at least latency between callbacks to allow
        # for accumulating changes
        now = Time.now.to_f
        diff = latency + (@last_queue_event_time || now) - now
        if diff > 0
          sleep diff
          next
        end

        _process_changes unless state == :paused
      end
    rescue RuntimeError
      Kernel.warn "[Listen warning]: Change block raised an exception: #{$!}"
      Kernel.warn "Backtrace:\n\t#{$@.join("\n\t")}"
    end

    def _silenced?(path, type)
      @silencer.silenced?(path, type)
    end

    def _start_adapter
      # Don't run async, because configuration has to finish first
      adapter = sync(:adapter)
      adapter.start
    end

    def _log(type, message)
      Celluloid.logger.send(type, message)
    end

    def _adapter_class
      @adapter_class ||= Adapter.select(options)
    end

    # for easier testing without sleep loop
    def _process_changes
      return if @queue.empty?

      @last_queue_event_time = nil

      changes = []
      while !@queue.empty?
        changes << @queue.pop
      end

      return if block.nil?

      hash = _smoosh_changes(changes)
      result = [hash[:modified], hash[:added], hash[:removed]]

      block_start = Time.now.to_f
      # TODO: condition not tested, but too complex to test ATM
      block.call(*result) unless result.all?(&:empty?)
      _log :debug, "Callback took #{Time.now.to_f - block_start} seconds"
    end

    attr_reader :wait_thread

    def _init_tcp_options(target)
      # Handle TCP options here
      require 'listen/tcp'
      fail ArgumentError, 'missing host/port for TCP' unless target

      if @tcp_mode == :recipient
        @options[:host] = 'localhost'
        @options[:force_tcp] = true
      end

      if target.is_a? Fixnum
        @options[:port] = target
        @port = target
      else
        @options[:host], port = target.split(':')
        @options[:port] = port.to_i
      end
    end

    def _reconfigure_silencer(extra_options)
      @options.merge!(extra_options)

      # TODO: this should be directory specific
      rules = [:only, :ignore, :ignore!].map do |option|
        [option, @options[option]] if @options.key? option
      end

      @silencer.configure(Hash[rules.compact])
    end

    def _start_wait_thread
      @wait_thread = Thread.new { _wait_for_changes }
    end

    def _wakeup_wait_thread
      wait_thread.wakeup if wait_thread && wait_thread.alive?
    end

    def _stop_wait_thread
      return unless wait_thread
      if wait_thread.alive?
        wait_thread.wakeup
        wait_thread.join
      end
      @wait_thread = nil
    end

    def _queue_raw_change(type, dir, rel_path, options)
      _log :debug, "raw queue: #{[type, dir, rel_path, options].inspect}"

      unless (worker = async(:change_pool))
        _log :warn, 'Failed to allocate worker from change pool'
        return
      end

      worker.change(type, dir, rel_path, options)
    rescue RuntimeError
      _log :error, "#{__method__} crashed: #{$!}:#{$@.join("\n")}"
      raise
    end
  end
end

1	require 'pathname'
2	require 'listen/adapter'
3	require 'listen/change'
4	require 'listen/record'
5	require 'listen/silencer'
6	require 'listen/queue_optimizer'
7	require 'English'
8
9	module Listen
10	class Listener
11	include Celluloid::FSM
12	include QueueOptimizer
13
14	attr_accessor :block
15
16	attr_reader :silencer
17
18	# TODO: deprecate
19	attr_reader :options, :directories
20	attr_reader :registry, :supervisor
21
22	# Initializes the directories listener.
23	#
24	# @param [String] directory the directories to listen to
25	# @param [Hash] options the listen options (see Listen::Listener::Options)
26	#
27	# @yield [modified, added, removed] the changed files
28	# @yieldparam [Array<String>] modified the list of modified files
29	# @yieldparam [Array<String>] added the list of added files
30	# @yieldparam [Array<String>] removed the list of removed files
31	#
32	def initialize(*args, &block)	334✔
33	@options = _init_options(args.last.is_a?(Hash) ? args.pop : {})	334✔
34
35	# Setup logging first
36	Celluloid.logger.level = _debug_level	334✔
37	_log :info, "Celluloid loglevel set to: #{Celluloid.logger.level}"	334✔
38
39	@silencer = Silencer.new	334✔
40	_reconfigure_silencer({})	334✔
41
42	@tcp_mode = nil	334✔
43	if [:recipient, :broadcaster].include? args[1]	168✔
44	target = args.shift	166✔
45	@tcp_mode = args.shift	166✔
46	_init_tcp_options(target)	165✔
47	end
48
49	@directories = args.flatten.map { \|path\| Pathname.new(path).realpath }	161✔
50	@queue = Queue.new	333✔
51	@block = block	333✔
52	@registry = Celluloid::Registry.new	333✔
53
54	transition :stopped	333✔
55	end
56
57	default_state :initializing
58
59	state :initializing, to: :stopped
60	state :paused, to: [:processing, :stopped]
61
62	state :stopped, to: [:processing] do	×
63	_stop_wait_thread	581✔
64	if @supervisor	333✔
65	@supervisor.terminate	248✔
66	@supervisor = nil	248✔
67	end
68	end
69
70	state :processing, to: [:paused, :stopped] do	×
71	if wait_thread # means - was paused	362✔
72	_wakeup_wait_thread	24✔
73	else
74	@last_queue_event_time = nil	338✔
75	_start_wait_thread	338✔
76	_init_actors	248✔
77
78	# Note: make sure building is finished before starting adapter (for
79	# consistent results both in specs and normal usage)
80	sync(:record).build	248✔
81
82	_start_adapter	248✔
83	end
84	end
85
86	# Starts processing events and starts adapters
87	# or resumes invoking callbacks if paused
88	def start	×
89	transition :processing	272✔
90	end
91
92	# TODO: depreciate
93	alias_method :unpause, :start
94
95	# Stops processing and terminates all actors
96	def stop	×
97	transition :stopped	412✔
98	end
99
100	# Stops invoking callbacks (messages pile up)
101	def pause	×
102	transition :paused	45✔
103	end
104
105	# processing means callbacks are called
106	def processing?	×
107	state == :processing	1✔
108	end
109
110	def paused?	×
111	state == :paused	1✔
112	end
113
114	# TODO: deprecate
115	alias_method :listen?, :processing?
116
117	# TODO: deprecate
118	def paused=(value)
119	transition value ? :paused : :processing
120	end
121
122	# TODO: deprecate
123	alias_method :paused, :paused?
124
125	# Add files and dirs to ignore on top of defaults
126	#
127	# (@see Listen::Silencer for default ignored files and dirs)
128	#
129	def ignore(regexps)	×
130	_reconfigure_silencer(ignore: [options[:ignore], regexps])	7✔
131	end
132
133	# Replace default ignore patterns with provided regexp
134	def ignore!(regexps)	×
135	_reconfigure_silencer(ignore: [], ignore!: regexps)	8✔
136	end
137
138	# Listen only to files and dirs matching regexp
139	def only(regexps)	×
140	_reconfigure_silencer(only: regexps)	1✔
141	end
142
143	def async(type)	×
144	proxy = sync(type)	724✔
145	proxy ? proxy.async : nil	38✔
146	end
147
148	def sync(type)	×
149	@registry[type]	1,614✔
150	end
151
152	def queue(type, change, dir, path, options = {})	31✔
153	fail "Invalid type: #{type.inspect}" unless [:dir, :file].include? type	390✔
154	fail "Invalid change: #{change.inspect}" unless change.is_a?(Symbol)	390✔
155	fail "Invalid path: #{path.inspect}" unless path.is_a?(String)	390✔
156	@queue << [type, change, dir, path, options]	390✔
157
158	@last_queue_event_time = Time.now.to_f	390✔
159	_wakeup_wait_thread unless state == :paused	283✔
160
161	return unless @tcp_mode == :broadcaster	175✔
162
163	message = TCP::Message.new(type, change, dir, path, options)	215✔
164	registry[:broadcaster].async.broadcast(message.payload)	215✔
165	end
166
167	private
168
169	def _init_options(options = {})	334✔
170	{ debug: false,	334✔
171	latency: nil,	334✔
172	wait_for_delay: 0.1,	334✔
173	force_polling: false,	334✔
174	polling_fallback_message: nil }.merge(options)	334✔
175	end
176
177	def _debug_level	×
178	debugging = ENV['LISTEN_GEM_DEBUGGING'] \|\| options[:debug]	334✔
179	case debugging.to_s	334✔
180	when /2/	334✔
181	Logger::DEBUG	×
182	when /true\|yes\|1/i	334✔
183	Logger::INFO	334✔
184	else
185	Logger::ERROR	334✔
186	end
187	end
188
189	def _init_actors	×
190	@supervisor = Celluloid::SupervisionGroup.run!(registry)	248✔
191	supervisor.add(Record, as: :record, args: self)	248✔
192	supervisor.pool(Change, as: :change_pool, args: self)	248✔
193
194	if @tcp_mode == :broadcaster	149✔
195	require 'listen/tcp/broadcaster'	99✔
196	supervisor.add(TCP::Broadcaster, as: :broadcaster, args: [@options[:host], @options[:port]])	99✔
197
198	# TODO: should be auto started, because if it crashes
199	# a new instance is spawned by supervisor, but it's 'start' isn't
200	# called
201	registry[:broadcaster].start	99✔
202	end
203
204	options = [mq: self, directories: directories]	248✔
205	supervisor.add(_adapter_class, as: :adapter, args: options)	248✔
206	end
207
208	def _wait_for_changes	×
209	latency = options[:wait_for_delay]	338✔
210
211	loop do	248✔
212	break if state == :stopped	821✔
213
214	if state == :paused \|\| @queue.empty?	281✔
215	sleep	450✔
216	break if state == :stopped	202✔
217	end
218
219	# Assure there's at least latency between callbacks to allow
220	# for accumulating changes
221	now = Time.now.to_f	483✔
222	diff = latency + (@last_queue_event_time \|\| now) - now	483✔
223	if diff > 0	196✔
224	sleep diff	287✔
225	next	287✔
226	end
227
228	_process_changes unless state == :paused	196✔
229	end
230	rescue RuntimeError	×
231	Kernel.warn "[Listen warning]: Change block raised an exception: #{$!}"	×
232	Kernel.warn "Backtrace:\n\t#{$@.join("\n\t")}"	×
233	end
234
235	def _silenced?(path, type)	×
236	@silencer.silenced?(path, type)	117✔
237	end
238
239	def _start_adapter	×
240	# Don't run async, because configuration has to finish first
241	adapter = sync(:adapter)	248✔
242	adapter.start	248✔
243	end
244
245	def _log(type, message)	×
246	Celluloid.logger.send(type, message)	1,152✔
247	end
248
249	def _adapter_class	×
250	@adapter_class \|\|= Adapter.select(options)	346✔
251	end
252
253	# for easier testing without sleep loop
254	def _process_changes	×
255	return if @queue.empty?	196✔
256
257	@last_queue_event_time = nil	196✔
258
259	changes = []	196✔
260	while !@queue.empty?	527✔
261	changes << @queue.pop	331✔
262	end
263
264	return if block.nil?	91✔
265
266	hash = _smoosh_changes(changes)	91✔
267	result = [hash[:modified], hash[:added], hash[:removed]]	91✔
268
269	block_start = Time.now.to_f	91✔
270	# TODO: condition not tested, but too complex to test ATM
271	block.call(*result) unless result.all?(&:empty?)	91✔
272	_log :debug, "Callback took #{Time.now.to_f - block_start} seconds"	91✔
273	end
274
275	attr_reader :wait_thread
276
277	def _init_tcp_options(target)	×
278	# Handle TCP options here
279	require 'listen/tcp'	166✔
280	fail ArgumentError, 'missing host/port for TCP' unless target	1✔
281
282	if @tcp_mode == :recipient	102✔
283	@options[:host] = 'localhost'	63✔
284	@options[:force_tcp] = true	63✔
285	end
286
287	if target.is_a? Fixnum	165✔
288	@options[:port] = target	165✔
289	@port = target	165✔
290	else
291	@options[:host], port = target.split(':')	×
292	@options[:port] = port.to_i	×
293	end
294	end
295
296	def _reconfigure_silencer(extra_options)	×
297	@options.merge!(extra_options)	350✔
298
299	# TODO: this should be directory specific
300	rules = [:only, :ignore, :ignore!].map do \|option\|	350✔
301	[option, @options[option]] if @options.key? option	985✔
302	end
303
304	@silencer.configure(Hash[rules.compact])	350✔
305	end
306
307	def _start_wait_thread	×
308	@wait_thread = Thread.new { _wait_for_changes }	248✔
309	end
310
311	def _wakeup_wait_thread	×
312	wait_thread.wakeup if wait_thread && wait_thread.alive?	2✔
313	end
314
315	def _stop_wait_thread	×
316	return unless wait_thread	333✔
317	if wait_thread.alive?	248✔
318	wait_thread.wakeup	248✔
319	wait_thread.join	248✔
320	end
321	@wait_thread = nil	248✔
322	end
323
324	def _queue_raw_change(type, dir, rel_path, options)	×
325	_log :debug, "raw queue: #{[type, dir, rel_path, options].inspect}"	561✔
326
327	unless (worker = async(:change_pool))	523✔
328	_log :warn, 'Failed to allocate worker from change pool'	38✔
329	return	38✔
330	end
331
332	worker.change(type, dir, rel_path, options)	523✔
333	rescue RuntimeError	×
334	_log :error, "#{__method__} crashed: #{$!}:#{$@.join("\n")}"	×
335	raise	×
336	end
337	end
338	end

guard / listen / 800

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