Class: ShadowsocksRuby::Connections::Connection

Inherits:
EventMachine::Connection
  • Object
show all
Defined in:
lib/shadowsocks_ruby/connections/connection.rb

Overview

This class added fiber enabled asynchronously receive function and pressure controled send_data to EventMachine::Connection

User code should define process_hook which hopefully implement a state machine .

Note: User code should not override post_init and receive_data, it is by design.

Examples:

class DummyConnection < ShadowsocksRuby::Connections::Connection
  def process_hook
    @i ||= 0
    @i += 1
    puts "I'm now in a fiber enabled context: #{@fiber}"
    Fiber.yield if @i >= 3
  end
end

Direct Known Subclasses

BackendConnection, ServerConnection

Constant Summary collapse

PressureLevel =

512K, used to pause plexer when plexer.get_outbound_data_size > this value

524288

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#logger ⇒ Object

get the logger object, the defautl logger is App.instance.logger



51
52
53
 
# File 'lib/shadowsocks_ruby/connections/connection.rb', line 51

def logger
  @logger ||= App.instance.logger
end

#plexer ⇒ Connection

It is where to relay peer’s traffic to For a server connection, plexer is backend connection. For a backend connection, plexer is server connection.

Returns:



45
46
47
 
# File 'lib/shadowsocks_ruby/connections/connection.rb', line 45

def plexer
  @plexer
end

Instance Method Details

#async_recv(n) ⇒ String

Asynchronously receive n bytes from @buffer

Parameters:

  • n (Integer) —

    Bytes to receive, if n = -1 returns all data in @buffer

Returns:

  • (String) —

    Returned n bytes data



94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/shadowsocks_ruby/connections/connection.rb', line 94

def async_recv n
  # wait n bytes
  if n == -1 && @buffer.size == 0 || @buffer.size < n
    @wait_length = n
    Fiber.yield
  end
  # read n bytes from buffer
  if n == -1
    s, @buffer = @buffer, String.new('', encoding: Encoding::ASCII_8BIT)
    return s
  else
    return @buffer.slice!(0, n)
  end
end

#async_recv_until(str) ⇒ String

Asynchronously receive data until str (eg: “\r\nr\n”) appears.

Parameters:

  • str (String) —

    Desired endding str

Returns:

  • (String) —

    Returned data, with str at end

Raises:

  • BufferOversizeError raise if cannot find str in first 65536 bytes (64K bytes)of @buffer, enough for a HTTP request head.



114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/shadowsocks_ruby/connections/connection.rb', line 114

def async_recv_until str
  # wait for str
  pos = @buffer =~ Regexp.new(str)
  while pos == nil
    @wait_length = -1
    Fiber.yield
    pos = @buffer =~ Regexp.new(str)
    raise BufferOversizeError, "oversized async_recv_until read" if @buffer.size > 65536
  end
  # read until str from buffer
  return @buffer.slice!(0, pos + str.length)
end

#peer ⇒ Object 



83
84
85
86
87
88
89
 
# File 'lib/shadowsocks_ruby/connections/connection.rb', line 83

def peer
  @peer ||=
  begin
    port, ip = Socket.unpack_sockaddr_in(get_peername)
    "#{ip}:#{port}"
  end
end

#send_data(data) ⇒ Object

send_data with pressure control @param data Data to send asynchronously



58
59
60
61
 
# File 'lib/shadowsocks_ruby/connections/connection.rb', line 58

def send_data data
  pressure_control
  super data
end

#unbind ⇒ Object

Close plexer first if it exists



171
172
173
174
 
# File 'lib/shadowsocks_ruby/connections/connection.rb', line 171

def unbind
  @plexer ||= nil
  @plexer.close_connection_after_writing if @plexer != nil
end