Class: Roma::Event::EMConPool

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/roma/event/con_pool.rb

Overview

module EMConnection

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ EMConPool

Returns a new instance of EMConPool.



84
85
86
87
88
89
 
# File 'lib/roma/event/con_pool.rb', line 84

def initialize
  @pool = {}
  @maxlength = 30
  @expire_time = 30
  @lock = Mutex.new
end

Instance Attribute Details

#expire_time ⇒ Object

Returns the value of attribute expire_time.



82
83
84
 
# File 'lib/roma/event/con_pool.rb', line 82

def expire_time
  @expire_time
end

#maxlength ⇒ Object

Returns the value of attribute maxlength.



81
82
83
 
# File 'lib/roma/event/con_pool.rb', line 81

def maxlength
  @maxlength
end

#pool ⇒ Object (readonly)

Returns the value of attribute pool.



80
81
82
 
# File 'lib/roma/event/con_pool.rb', line 80

def pool
  @pool
end

Instance Method Details

#close_all ⇒ Object 



125
126
127
 
# File 'lib/roma/event/con_pool.rb', line 125

def close_all
  @pool.each_key{|ap| close_at(ap) }
end

#close_at(ap) ⇒ Object 



136
137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/roma/event/con_pool.rb', line 136

def close_at(ap)
  return unless @pool.key?(ap)
  @lock.synchronize {
    while(@pool[ap].length > 0)
      begin
        @pool[ap].shift.close_connection
      rescue =>e
        Roma::Logging::RLogger.instance.error("#{__FILE__}:#{__LINE__}:#{e.inspect}")
      end
    end
    @pool.delete(ap)
  }
end

#close_same_host(ap) ⇒ Object 



129
130
131
132
133
134
 
# File 'lib/roma/event/con_pool.rb', line 129

def close_same_host(ap)
  host,port = ap.split(/[:_]/)
  @pool.each_key{|eap|
    close_at(eap) if eap.split(/[:_]/)[0] == host
  }
end

#create_connection(ap) ⇒ Object 



117
118
119
120
121
122
123
 
# File 'lib/roma/event/con_pool.rb', line 117

def create_connection(ap)
  host,port = ap.split(/[:_]/)
  addr = DNSCache.instance.resolve_name(host)
  con = EventMachine::connect(addr, port, Roma::Event::EMConnection)
  con.ap = ap
  con
end

#get_connection(ap) ⇒ Object 



91
92
93
94
95
96
97
98
99
100
 
# File 'lib/roma/event/con_pool.rb', line 91

def get_connection(ap)
  ret = @pool[ap].shift if @pool.key?(ap) && @pool[ap].length > 0
  if ret && @expire_time != 0 && ret.last_access < Time.now - @expire_time
    ret.close_connection if ret.connected
    ret = nil
    Logging::RLogger.instance.info("EM connection expired at #{ap},remains #{@pool[ap].length}")
  end
  ret = create_connection(ap) if ret == nil || ret.connected != true
  ret
end

#return_connection(ap, con) ⇒ Object 



102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/roma/event/con_pool.rb', line 102

def return_connection(ap, con)
  return if con.connected == false

  con.last_access = Time.now
  if @pool.key?(ap) && @pool[ap].length > 0
    if @pool[ap].length > @maxlength
      con.close_connection
    else
      @pool[ap] << con
    end
  else
    @pool[ap] = [con]
  end
end