Class: ASIR::Transport::ConnectionOriented

Inherits:
Stream show all
Includes:
PayloadIO, UriConfig
Defined in:
lib/asir/transport/connection_oriented.rb

Overview

!SLIDE Connection-Oriented Transport

Direct Known Subclasses

TcpSocket

Constant Summary

Constants included from PayloadIO

PayloadIO::FOOTER, PayloadIO::HEADER

Constants included from UriConfig

UriConfig::S_LOCALHOST, UriConfig::S_TCP

Constants included from ASIR::ThreadVariable

ASIR::ThreadVariable::DEBUG, ASIR::ThreadVariable::EMPTY_HASH, ASIR::ThreadVariable::SETTER

Instance Attribute Summary

Attributes included from UriConfig

#host, #host_default, #path, #path_default, #port, #port_default, #scheme, #scheme_default, #uri

Attributes inherited from ASIR::Transport

#after_invoke_message, #after_receive_message, #before_send_message, #coder_needs_result_message, #decoder, #encoder, #invoker, #message_count, #needs_message_identifier, #needs_message_timestamp, #on_exception, #on_result_exception, #one_way, #running, #verbose

Attributes included from Log

#_logger

Instance Method Summary collapse

Methods included from PayloadIO

#_read, #_read_line_and_expect!, #_write, #close

Methods included from UriConfig

#_uri, #address, #protocol

Methods inherited from Stream

#_serve_stream!, #serve_stream!, #serve_stream_message!, #stream_eof?

Methods inherited from ASIR::Transport

#_subclass_responsibility, #initialize, #invoke_message!, #needs_message_identifier?, #needs_message_timestamp?, #receive_message, #receive_result, #send_message, #send_result, #serve_message!, #stop!, #with_server_signals!

Methods included from Log

#_log, #_log_enabled=, #_log_enabled?, #_log_format, #_log_result, enabled, enabled=, included

Methods included from Initialization

#initialize

Methods included from AdditionalData

#[], #[]=, #_additional_data, #additional_data, #additional_data!, #additional_data=, included

Methods included from Message::Delay

#relative_message_delay!, #wait_for_delay!

Methods included from ASIR::ThreadVariable

included, setter

Constructor Details

This class inherits a constructor from ASIR::Transport

Instance Method Details

#_after_connect!(stream) ⇒ Object

Subclasses can override.



44
45
46
 
# File 'lib/asir/transport/connection_oriented.rb', line 44

def _after_connect! stream
  self
end

#_before_close!(stream) ⇒ Object

Subclasses can override.



49
50
51
 
# File 'lib/asir/transport/connection_oriented.rb', line 49

def _before_close! stream
  self
end

#_connect! ⇒ Object

Returns raw client stream.



33
34
35
36
37
38
39
40
41
 
# File 'lib/asir/transport/connection_oriented.rb', line 33

def _connect!
  _log { "_connect! #{uri}" } if @verbose >= 1
  stream = _client_connect!
  _log { "_connect! stream=#{stream}" } if @verbose >= 1
  _after_connect! stream
  stream
rescue ::Exception => err
  raise err.class, "Cannot connect to #{self.class} #{uri}: #{err.inspect}", err.backtrace
end

#_receive_message(state) ⇒ Object

!SLIDE Receives the encoded Message payload String.



64
65
66
 
# File 'lib/asir/transport/connection_oriented.rb', line 64

def _receive_message state
  state.message_payload = _read(state.in_stream, state)
end

#_receive_result(state) ⇒ Object

!SLIDE Receives the encoded Result payload String.



80
81
82
83
84
85
 
# File 'lib/asir/transport/connection_oriented.rb', line 80

def _receive_result state
  unless @one_way || state.message.one_way
    state.result_payload = _read(state.in_stream, state)
    true
  end
end

#_send_message(state) ⇒ Object

!SLIDE Sends the encoded Message payload String.



55
56
57
58
59
60
 
# File 'lib/asir/transport/connection_oriented.rb', line 55

def _send_message state
  stream.with_stream! do | io |
    state.in_stream = io
    _write(state.message_payload, io, state)
  end
end

#_send_result(state) ⇒ Object

!SLIDE Sends the encoded Result payload String.



70
71
72
73
74
75
76
 
# File 'lib/asir/transport/connection_oriented.rb', line 70

def _send_result state
  unless @one_way || state.message.one_way
    # $stderr.write "\n  _send_result #{state.result_payload.inspect}\n\n"
    _write(state.result_payload, state.out_stream, state)
    true
  end
end

#_server! ⇒ Object

Raises:



133
134
135
 
# File 'lib/asir/transport/connection_oriented.rb', line 133

def _server!
  raise Error::SubclassResponsibility, "_server!"
end

#_server_accept_connection!(server) ⇒ Object

Accept a client connection. Returns [ in_stream, out_stream ].

Raises:



147
148
149
 
# File 'lib/asir/transport/connection_oriented.rb', line 147

def _server_accept_connection! server
  raise Error::SubclassResponsibility, "_server_accept_connection!"
end

#_server_close! ⇒ Object 



137
138
139
140
141
142
143
 
# File 'lib/asir/transport/connection_oriented.rb', line 137

def _server_close!
  if @server
    @server.close rescue nil
  end
  @server = nil
  self
end

#_server_close_connection!(in_stream, out_stream) ⇒ Object

Close a client connection.

Raises:



152
153
154
 
# File 'lib/asir/transport/connection_oriented.rb', line 152

def _server_close_connection! in_stream, out_stream
  raise Error::SubclassResponsibility, "_server_close_connection!"
end

#connect!(opts = nil, &blk) ⇒ Object

Yields Channel after _connect!.



20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/asir/transport/connection_oriented.rb', line 20

def connect!(opts = nil, &blk)
  base_opts = {
    :on_connect => lambda { | channel |
      connection = _connect!
      blk.call(connection) if blk
      connection
    }
  }
  base_opts.update(opts) if opts
  Channel.new(base_opts)
end

#prepare_server! ⇒ Object

!SLIDE Server



90
91
92
93
94
95
96
 
# File 'lib/asir/transport/connection_oriented.rb', line 90

def prepare_server!
  _log { "prepare_server! #{uri}" } if @verbose >= 1
  _server!
rescue ::Exception => err
  _log [ "prepare_server! #{uri}", :exception, err ]
  raise err.class, "Cannot prepare server on #{self.class} #{uri}: #{err.inspect}", err.backtrace
end

#run_server! ⇒ Object 



98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/asir/transport/connection_oriented.rb', line 98

def run_server!
  _log { "run_server! #{uri}" } if @verbose >= 1
  with_server_signals! do
    @running = true
    server_on_start!
    while @running
      serve_connection!
    end
  end
  self
ensure
  server_on_stop!
  _server_close!
end

#serve_connection! ⇒ Object 



119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/asir/transport/connection_oriented.rb', line 119

def serve_connection!
  _log { "serve_connection!: accepting connection" } if @verbose >= 2
  in_stream, out_stream = _server_accept_connection! @server
  _log { "serve_connection!: connected #{in_stream} #{out_stream}" } if @verbose >= 1
  _server_serve_stream! in_stream, out_stream
rescue Error::Terminate => err
  @running = false
  _log [ :serve_connection_terminate, err ]
ensure
  _server_close_connection!(in_stream, out_stream)
  _log { "serve_connection!: disconnected" } if @verbose >= 1
end

#server_on_start! ⇒ Object 



113
114
 
# File 'lib/asir/transport/connection_oriented.rb', line 113

def server_on_start!
end

#server_on_stop! ⇒ Object 



116
117
 
# File 'lib/asir/transport/connection_oriented.rb', line 116

def server_on_stop!
end

#stream ⇒ Object

!SLIDE Returns a connected Channel.



14
15
16
17
 
# File 'lib/asir/transport/connection_oriented.rb', line 14

def stream
  @stream ||=
    connect!
end