Class: AlWorker::BroadcastMessage

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

Overview

Broadcast message

(note) スレッド間メッセージングシステム。1:nメッセージを実現する。（通常の1:1メッセージであれば、Ruby標準のQueueで十分。）

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ BroadcastMessage

constructor



33
34
35
 
# File 'lib/al_worker_message.rb', line 33

def initialize()
  @threads = {}
end

Instance Attribute Details

#threads ⇒ Hash <Thread,Queue> (readonly)

Returns スレッドとメッセージキュー.

Returns:

  • (Hash <Thread,Queue>) —

    スレッドとメッセージキュー



27
28
29
 
# File 'lib/al_worker_message.rb', line 27

def threads
  @threads
end

Instance Method Details

#attach ⇒ Object

ブロードキャストメッセージ受信予約



41
42
43
44
 
# File 'lib/al_worker_message.rb', line 41

def attach()
  AlWorker.log("Attach #{Thread.current.object_id}", :debug, "BCM" )
  @threads[ Thread.current.object_id ] = [ Thread.current, Queue.new() ]
end

#detach ⇒ Object

ブロードキャストメッセージ受信予約解除



50
51
52
53
 
# File 'lib/al_worker_message.rb', line 50

def detach()
  AlWorker.log("Detach #{Thread.current.object_id}", :debug, "BCM" )
  @threads.delete( Thread.current.object_id )
end

#empty? ⇒ Boolean

メッセージがあるか問い合わせ

Returns:

  • (Boolean) —

    メッセージが無い時、true。



92
93
94
95
96
 
# File 'lib/al_worker_message.rb', line 92

def empty?()
  @threads[ Thread.current.object_id ][1].empty?()
rescue NoMethodError
  raise "Maybe used empty?() without attach()."
end

#receive ⇒ Object

Note:

メッセージ受信

メッセージがなければ、送られるまで停止する。

Returns:

  • (Object) —

    受信メッセージ



80
81
82
83
84
 
# File 'lib/al_worker_message.rb', line 80

def receive()
  @threads[ Thread.current.object_id ][1].pop()
rescue NoMethodError
  raise "Maybe used receive() without attach()."
end

#send(msg) ⇒ Object

メッセージ送信

Parameters:

  • msg (Object) —

    送信メッセージ



61
62
63
64
65
66
67
68
69
70
 
# File 'lib/al_worker_message.rb', line 61

def send( msg )
  @threads.values.each do |th,q|
    if th.alive?
      q.push( msg )
    else
      AlWorker.log("detach #{th.object_id}", :debug, "BCM" )
      @threads.delete( th.object_id )
    end
  end
end