Class: Thread::Channel

Inherits:
Object
  • Object
show all
Defined in:
lib/thread/channel.rb

Overview

A channel lets you send and receive various messages in a thread-safe way.

It also allows for guards upon sending and retrieval, to ensure the passed messages are safe to be consumed.

Instance Method Summary collapse

Constructor Details

#initialize(messages = [], &block) ⇒ Channel

Create a channel with optional initial messages and optional channel guard.



19
20
21
22
23
24
25
26
27
 
# File 'lib/thread/channel.rb', line 19

def initialize (messages = [], &block)
	@messages = []
	@mutex    = Mutex.new
	@check    = block

	messages.each {|o|
		send o
	}
end

Instance Method Details

#receive(&block) ⇒ Object

Receive a message, if there are none the call blocks until there’s one.

If a block is passed, it’s used as guard to match to a message.



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/thread/channel.rb', line 50

def receive (&block)
	message = nil
	found   = false

	if block
		until found
			@mutex.synchronize {
				if index = @messages.find_index(&block)
					message = @messages.delete_at(index)
					found   = true
				else
					cond.wait @mutex
				end
			}
		end
	else
		until found
			@mutex.synchronize {
				if @messages.empty?
					cond.wait @mutex
				end

				unless @messages.empty?
					message = @messages.shift
					found   = true
				end
			}
		end
	end

	message
end

#receive!(&block) ⇒ Object

Receive a message, if there are none the call returns nil.

If a block is passed, it’s used as guard to match to a message.



86
87
88
89
90
91
92
 
# File 'lib/thread/channel.rb', line 86

def receive! (&block)
	if block
		@messages.delete_at(@messages.find_index(&block))
	else
		@messages.shift
	end
end

#send(what) ⇒ Object

Send a message to the channel.

If there’s a guard, the value is passed to it, if the guard returns a falsy value an ArgumentError exception is raised and the message is not sent.



33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/thread/channel.rb', line 33

def send (what)
	if @check && !@check.call(what)
		raise ArgumentError, 'guard mismatch'
	end

	@mutex.synchronize {
		@messages << what

		cond.broadcast if cond?
	}

	self
end