Module: Hakuban::Stream

Included in:
ObjectExposeContract, ObjectObserveContract, ObjectStateSink, ObjectStateStream, TagExposeContract, TagObserveContract
Defined in:
lib/hakuban/stream.rb

Defined Under Namespace

Classes: NextItemInterrupt

Instance Method Summary collapse

Instance Method Details

#each(*args, **kwargs, &block) ⇒ Object 



50
51
52
 
# File 'lib/hakuban/stream.rb', line 50

def each(*args, **kwargs, &block)
	for_each(*args, **kwargs, &block)
end

#for_each(*args, **kwargs, &block) ⇒ Object 



54
55
56
57
58
59
60
 
# File 'lib/hakuban/stream.rb', line 54

def for_each(*args, **kwargs, &block)
	if kwargs[:interrupt_on_next]
		for_each_till_next(*args, **kwargs, &block)
	else
		while self.next { |new_item| sync_call(new_item, args, kwargs, block); true }; end
	end
end

#for_each_concurrent(*args, **kwargs, &block) ⇒ Object 



68
69
70
 
# File 'lib/hakuban/stream.rb', line 68

def for_each_concurrent(*args, **kwargs, &block)
	for_each_in_thread(*args, **kwargs, &block)
end

#for_each_in_thread(*args, **kwargs, &block) ⇒ Object 



73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/hakuban/stream.rb', line 73

def for_each_in_thread(*args, **kwargs, &block)			
	Thread.handle_interrupt(Object => :never) {

		# we poll the stream in a separate thread because we sometimes want to interrupt the polling thread from an item-handling thread, and, without the wrapper-thread, we could end up raising exception in a random place of code, if Async is being used. maybe?
		stream_polling_thread = Thread.new {
			begin
				item_threads = ObjectSpace::WeakMap.new
				exception_to_rescue = NextItemInterrupt.new

				while new_item = self.next
					if kwargs[:kill_previous_on_next]
						item_threads.keys.each { |item_thread| 
							item_thread.raise(exception_to_rescue)
							item_thread.join
						}
					end

					item_threads[Thread.new(new_item, Thread.current) do |item, parent_thread|
						sync_call(item, args, kwargs, block)
					rescue Object => error
						if error != exception_to_rescue
							if kwargs[:propagate_exceptions].nil? or kwargs[:propagate_exceptions]
								parent_thread.raise(error)
							else
								raise
							end
						end
					ensure
						item_threads.delete Thread.current
					end] = true
				end

				if kwargs[:kill_previous_on_next]
					item_threads.keys.each { |item_thread| 
						item_thread.raise(exception_to_rescue)
						item_thread.join
					}
				end

				Thread.handle_interrupt(Object => :immediate) {
					item_threads.keys.each { |item_thread|
						item_thread.join
					}
				}

				nil
			rescue Object => error
				exception_to_rescue = error
				item_threads.keys.each { |item_thread| item_thread.raise(error) }
				error
			ensure
				item_threads.keys.each { |item_thread|
					begin
						item_thread.join_with_warning(60)
					rescue Object => error
						# it's probably better to not re-raise here, so this one doesn't obscure the original. but lets at least print it out.
						$stderr.puts "Item thread exception: \n"+error.inspect
					end
				}	
			end
		}
		 
		begin
			Thread.handle_interrupt(Object => :immediate) {
				value = stream_polling_thread.join.value
				raise value  if !value.nil?
			}
		rescue Object => error
			stream_polling_thread.raise(error)
			raise
		ensure
			begin
				stream_polling_thread.join_with_warning(60)
			rescue Object => error
				# it's probably better to not re-raise here, so this one doesn't obscure the original. but lets at least print it out.
				$stderr.puts "Item thread exception: \n"+error.inspect
			end
		end

	}
end

#for_each_till_next(*args, **kwargs, &block) ⇒ Object 



63
64
65
 
# File 'lib/hakuban/stream.rb', line 63

def for_each_till_next(*args, **kwargs, &block)
	for_each_in_thread(*args,**kwargs.merge(kill_previous_on_next: true), &block)			
end