Class: Baykit::BayServer::Agent::Multiplexer::JobMultiplexer

Inherits:
JobMultiplexerBase show all
Includes:
Baykit::BayServer::Agent::Multiplexer, TimerHandler, Common, Common::Recipient, Rudders, Util
Defined in:
lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb

Instance Attribute Summary

Attributes inherited from JobMultiplexerBase

#anchorable, #pipe

Attributes inherited from MultiplexerBase

#agent, #channel_count, #lock, #rudders, #rudders_lock

Instance Method Summary collapse

Methods included from Common::Recipient

#receive, #wakeup

Methods included from TimerHandler

#on_timer

Methods inherited from JobMultiplexerBase

#on_busy, #on_free, #on_timer, #shutdown

Methods inherited from MultiplexerBase

#add_rudder_state, #close_all, #close_rudder, #close_timeout_sockets, #consume_oldest_unit, #find_rudder_state_by_key, #get_rudder_state, #get_transporter, #is_busy, #remove_rudder_state

Methods included from Common::Multiplexer

#add_rudder_state, #close_rudder, #consume_oldest_unit, #get_rudder_state, #get_transporter, #is_busy, #on_busy, #on_free, #remove_rudder_state, #req_end, #shutdown

Constructor Details

#initialize(agt, anchorable) ⇒ JobMultiplexer

Returns a new instance of JobMultiplexer.



20
21
22
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 20

def initialize(agt, anchorable)
  super
end

Instance Method Details

#cancel_read(st) ⇒ Object 



180
181
182
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 180

def cancel_read(st)

end

#cancel_write(st) ⇒ Object 



184
185
186
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 184

def cancel_write(st)

end

#is_non_blocking ⇒ Object 



261
262
263
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 261

def is_non_blocking()
  return false
end

#next_accept(st) ⇒ Object 



188
189
190
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 188

def next_accept(st)
  req_accept(st.rudder)
end

#next_read(st) ⇒ Object 



192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 192

def next_read(st)
  id = st.id

  Thread.new do

    begin
      if st.handshaking
        # Calls accept API for client socket
        st.rudder.io.accept
        st.handshaking = false

        BayLog.debug("%s Handshake done (rd=%s)", self, st.rudder)
        app_protocols = st.rudder.io.context.alpn_protocols

        # HELP ME
        #   This code does not work!
        #   We cannot get application protocol name
        proto = nil
        if app_protocols != nil && app_protocols.length > 0
          proto = app_protocols[0]
        end
      end

      BayLog.debug("%s Try to Read (rd=%s)", @agent, st.rudder)
      n = st.rudder.read(st.read_buf, st.buf_size)

      if get_rudder_state(st.rudder) == nil
        #channel is already closed
        BayLog.debug("%s Rudder is already closed: rd=%s", self, st.rudder);
        next
      else
        @agent.send_read_letter(id, st.rudder, self, n, nil, true)
      end

    rescue Exception => e
      @agent.send_error_letter(id, st.rudder, self, e, true)
    end
  end
end

#next_write(st) ⇒ Object 



232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 232

def next_write(st)
  id = st.id

  Thread.new do
    BayLog.debug("%s next write st=%s", @agent, st)

    if st == nil
      BayLog.warn("%s Channel is closed: %s", @agent, st.rudder)
      next
    end

    u = st.write_queue[0]
    BayLog.debug("%s Try to write: pkt=%s buflen=%d", self, u.tag, u.buf.length)

    n = 0
    begin
      if u.buf.length > 0
        n = st.rudder.write(u.buf)
        u.buf.slice!(0, n)
      end
    rescue Exception => e
      @agent.send_error_letter(id, st.rudder, self, e, true)
      next
    end

    @agent.send_wrote_letter(id, st.rudder, self, n, true)
  end
end

#req_accept(rd) ⇒ Object

Implements Multiplexer



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 32

def req_accept(rd)
  BayLog.debug("%s reqAccept isShutdown=%s", @agent, @agent.aborted)
  if @agent.aborted
    return
  end

  st = get_rudder_state(rd)
  id = st.id

  Thread.new do
    begin
      if @agent.aborted
        next
      end

      begin
        client_skt, adr = rd.io.accept
      rescue Exception => e
        @agent.send_error_letter(id, rd, self, e, true)
        next
      end

      BayLog.debug("%s Accepted skt=%s", @agent, client_skt)
      if agent.aborted
        BayLog.error("%s Agent is not alive (close)", @agent);
        client_skt.close
      else
        @agent.send_accepted_letter(id, rd, self, IORudder.new(client_skt), true)
      end

    rescue Exception => e
      BayLog.fatal_e(e)
      @agent.shutdown
    end
  end

end

#req_close(rd) ⇒ Object 



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 150

def req_close(rd)
  st = get_rudder_state(rd)
  BayLog.debug("%s reqClose st=%s", @agent, st);

  if st == nil
    BayLog.warn("%s channel state not found: %s", @agent, rd)
    return
  end

  id = st.id

    Thread.new do
    begin
      st = get_rudder_state(rd)
      if st == nil
        BayLog.debug("%s Rudder is already closed: rd=%s", @agent, rd)
        next
      end

      close_rudder(rd)
      @agent.send_closed_letter(id, rd, self, true)
    rescue Exception => e
      BayLog.fatal_e(e)
      @agent.shutdown
    end
  end

  st.access
end

#req_connect(rd, adr) ⇒ Object 



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 71

def req_connect(rd, adr)
  BayLog.debug("%s reqConnect adr=%s rd=%s", @agent, adr.canonname, rd)

  Thread.new do
    st = get_rudder_state(rd)
    if st == nil
      # rudder is already closed
      BayLog.debug("%s Rudder is already closed: rd=%s", @agent, rd)
      return
    end

    begin
      rd.io.connect(adr)
      BayLog.debug("%s Connected rd=%s", @agent, rd)
      @agent.send_connected_letter(st.id, rd, self, true)
    rescue Exception => e
      @agent.send_error_letter(st.id, rd, self, e, true)
      return
    end
  end

  st = get_rudder_state(rd)
  st.access
  st.connecting = true
end

#req_read(rd) ⇒ Object 



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 97

def req_read(rd)
  st = get_rudder_state(rd)
  if st == nil
    return
  end

  BayLog.debug("%s reqRead rd=%s state=%s", @agent, st.rudder, st);
  need_read = false
  st.reading_lock.synchronize do
    if !st.reading
      need_read = true
      st.reading = true
    end
  end

  if need_read
    next_read(st)
  end

  st.access
end

#req_write(rd, buf, adr, tag, &lis) ⇒ Object 



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
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 119

def req_write(rd, buf, adr, tag, &lis)
  st = get_rudder_state(rd)
  BayLog.debug("%s reqWrite st=%s", @agent, st)

  if st == nil
    raise IOError.new("Invalid rudder")
    #BayLog.warn("%s Channel is closed(callback immediately): %s", @agent, rd)
    #lis.call()
    #return
  end

  unt = WriteUnit.new(buf, adr, tag, &lis)
  st.write_queue_lock.synchronize do
    st.write_queue << unt
  end

  need_write = false
  st.writing_lock.synchronize do
    if !st.writing
      need_write = true
      st.writing = true
    end
  end

  if need_write
    next_write(st)
  end

  st.access
end

#to_s ⇒ Object 



23
24
25
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 23

def to_s
  return "JobMpx[#{@agent}]"
end

#use_async_api ⇒ Object 



265
266
267
 
# File 'lib/baykit/bayserver/agent/multiplexer/job_multiplexer.rb', line 265

def use_async_api()
  return false
end