Class: Vines::Stream

Inherits:
EventMachine::Connection
  • Object
show all
Includes:
Log
Defined in:
lib/vines/stream.rb,
lib/vines/stream/http.rb,
lib/vines/stream/sasl.rb,
lib/vines/stream/state.rb,
lib/vines/stream/client.rb,
lib/vines/stream/parser.rb,
lib/vines/stream/server.rb,
lib/vines/stream/component.rb,
lib/vines/stream/http/auth.rb,
lib/vines/stream/http/bind.rb,
lib/vines/stream/client/tls.rb,
lib/vines/stream/http/ready.rb,
lib/vines/stream/http/start.rb,
lib/vines/stream/server/tls.rb,
lib/vines/stream/client/auth.rb,
lib/vines/stream/client/bind.rb,
lib/vines/stream/server/auth.rb,
lib/vines/stream/client/ready.rb,
lib/vines/stream/client/start.rb,
lib/vines/stream/http/request.rb,
lib/vines/stream/http/session.rb,
lib/vines/stream/server/ready.rb,
lib/vines/stream/server/start.rb,
lib/vines/stream/client/closed.rb,
lib/vines/stream/http/sessions.rb,
lib/vines/stream/client/session.rb,
lib/vines/stream/component/ready.rb,
lib/vines/stream/component/start.rb,
lib/vines/stream/http/bind_restart.rb,
lib/vines/stream/client/auth_restart.rb,
lib/vines/stream/client/bind_restart.rb,
lib/vines/stream/component/handshake.rb,
lib/vines/stream/server/auth_restart.rb,
lib/vines/stream/server/outbound/tls.rb,
lib/vines/stream/server/final_restart.rb,
lib/vines/stream/server/outbound/auth.rb,
lib/vines/stream/server/outbound/start.rb,
lib/vines/stream/server/outbound/tls_result.rb,
lib/vines/stream/server/outbound/auth_result.rb,
lib/vines/stream/server/outbound/auth_restart.rb,
lib/vines/stream/server/outbound/final_restart.rb,
lib/vines/stream/server/outbound/final_features.rb

Overview

The base class for various XMPP streams (c2s, s2s, component, http), containing behavior common to all streams like rate limiting, stanza parsing, and stream error handling.

Direct Known Subclasses

Client, Component, Server

Defined Under Namespace

Classes: Client, Component, Http, Parser, SASL, Server, State

Constant Summary collapse

ERROR = 
'error'.freeze
PAD = 
20

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Log

#log

Constructor Details

#initialize(config) ⇒ Stream

Returns a new instance of Stream.



16
17
18
 
# File 'lib/vines/stream.rb', line 16

def initialize(config)
  @config = config
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



13
14
15
 
# File 'lib/vines/stream.rb', line 13

def config
  @config
end

#domain ⇒ Object (readonly)

Returns the value of attribute domain.



13
14
15
 
# File 'lib/vines/stream.rb', line 13

def domain
  @domain
end

#user ⇒ Object

Returns the value of attribute user.



14
15
16
 
# File 'lib/vines/stream.rb', line 14

def user
  @user
end

Instance Method Details

#advance(state) ⇒ Object

Advance the stream’s state machine to the new state. XML nodes received by the stream will be passed to this state’s ‘node` method.

state - The Stream::State to process the stanzas next.

Returns the new Stream::State.



173
174
175
 
# File 'lib/vines/stream.rb', line 173

def advance(state)
  @state = state
end

#available_resources(*jid) ⇒ Object 



115
116
117
 
# File 'lib/vines/stream.rb', line 115

def available_resources(*jid)
  router.available_resources(*jid, user.jid)
end

#cert_domain_matches?(domain) ⇒ Boolean

Returns:

  • (Boolean) 


132
133
134
 
# File 'lib/vines/stream.rb', line 132

def cert_domain_matches?(domain)
  @store.domain?(get_peer_cert, domain)
end

#close_connection(after_writing = false) ⇒ Object

Advance the state machine into the ‘Closed` state so any remaining queued nodes are not processed while we’re waiting for EM to actually close the connection.

Returns nothing.



56
57
58
59
60
 
# File 'lib/vines/stream.rb', line 56

def close_connection(after_writing=false)
  super
  @closed = true
  advance(Client::Closed.new(self))
end

#connected_resources(jid) ⇒ Object 



111
112
113
 
# File 'lib/vines/stream.rb', line 111

def connected_resources(jid)
  router.connected_resources(jid, user.jid)
end

#create_parser ⇒ Object

Initialize a new XML parser for this connection. This is called when the stream is first connected as well as for stream restarts during negotiation. Subclasses can override this method to provide a different type of parser (e.g. HTTP).

Returns nothing.



43
44
45
46
47
48
49
 
# File 'lib/vines/stream.rb', line 43

def create_parser
  @parser = Parser.new.tap do |parser|
    parser.stream_open {|node| @nodes.push(node) }
    parser.stream_close { close_connection }
    parser.stanza {|node| @nodes.push(node) }
  end
end

#encrypt ⇒ Object 



149
150
151
152
 
# File 'lib/vines/stream.rb', line 149

def encrypt
  cert, key = @store.files_for_domain(domain)
  start_tls(cert_chain_file: cert, private_key_file: key, verify_peer: true)
end

#encrypt? ⇒ Boolean

Returns true if the TLS certificate and private key files for this domain exist and can be used to encrypt this stream.

Returns:

  • (Boolean) 


156
157
158
 
# File 'lib/vines/stream.rb', line 156

def encrypt?
  !@store.files_for_domain(domain).nil?
end

#error(e) ⇒ Object

Stream level errors close the stream while stanza and SASL errors are written to the client and leave the stream open. All exceptions should pass through this method for consistent handling.

e - The StandardError, usually XmppError, that occurred.

Returns nothing.



184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/vines/stream.rb', line 184

def error(e)
  case e
  when SaslError, StanzaError
    write(e.to_xml)
  when StreamError
    send_stream_error(e)
    close_stream
  else
    log.error(e)
    send_stream_error(StreamErrors::InternalServerError.new)
    close_stream
  end
end

#interested_resources(*jid) ⇒ Object 



119
120
121
 
# File 'lib/vines/stream.rb', line 119

def interested_resources(*jid)
  router.interested_resources(*jid, user.jid)
end

#post_init ⇒ Object

Initialize the stream after its connection to the server has completed. EventMachine calls this method when an incoming connection is accepted into the event loop.

Returns nothing.



25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/vines/stream.rb', line 25

def post_init
  @remote_addr, @local_addr = addresses
  @user, @closed, @stanza_size = nil, false, 0
  @bucket = TokenBucket.new(100, 10)
  @store = Store.new(@config.certs)
  @nodes = EM::Queue.new
  process_node_queue
  create_parser
  log.info { "%s %21s -> %s" %
    ['Stream connected:'.ljust(PAD), @remote_addr, @local_addr] }
end

#receive_data(data) ⇒ Object

Read bytes off the stream and feed them into the XML parser. EventMachine is responsible for calling this method on its event loop as connections become readable.

data - The byte String sent to the server from the client, hopefully XML.

Returns nothing.



69
70
71
72
73
74
75
76
77
 
# File 'lib/vines/stream.rb', line 69

def receive_data(data)
  return if @closed
  @stanza_size += data.bytesize
  if @stanza_size < max_stanza_size
    @parser << data rescue error(StreamErrors::NotWellFormed.new)
  else
    error(StreamErrors::PolicyViolation.new('max stanza size reached'))
  end
end

#reset ⇒ Object

Reset the connection’s XML parser when a new <stream:stream> header is received.

Returns nothing.



83
84
85
 
# File 'lib/vines/stream.rb', line 83

def reset
  create_parser
end

#router ⇒ Object 



198
199
200
 
# File 'lib/vines/stream.rb', line 198

def router
  @config.router
end

#ssl_verify_peer(pem) ⇒ Object 



123
124
125
126
127
128
129
130
 
# File 'lib/vines/stream.rb', line 123

def ssl_verify_peer(pem)
  # EM is supposed to close the connection when this returns false,
  # but it only does that for inbound connections, not when we
  # make a connection to another server.
  @store.trusted?(pem).tap do |trusted|
    close_connection unless trusted
  end
end

#storage(domain = nil) ⇒ Object

Returns the storage system for the domain. If no domain is given, the stream’s storage mechanism is returned.



89
90
91
 
# File 'lib/vines/stream.rb', line 89

def storage(domain=nil)
  @config.storage(domain || self.domain)
end

#unbind ⇒ Object 



160
161
162
163
164
165
 
# File 'lib/vines/stream.rb', line 160

def unbind
  router.delete(self)
  log.info { "%s %21s -> %s" %
    ['Stream disconnected:'.ljust(PAD), @remote_addr, @local_addr] }
  log.info { "Streams connected: #{router.size}" }
end

#update_user_streams(user) ⇒ Object

Reload the user’s information into their active connections. Call this after storage.save_user() to sync the new user state with their other connections.

user - The User whose connection info needs refreshing.

Returns nothing.



105
106
107
108
109
 
# File 'lib/vines/stream.rb', line 105

def update_user_streams(user)
  connected_resources(user.jid.bare).each do |stream|
    stream.user.update_from(user)
  end
end

#vhost ⇒ Object

Returns the Config::Host virtual host for the stream’s domain.



94
95
96
 
# File 'lib/vines/stream.rb', line 94

def vhost
  @config.vhost(domain)
end

#write(data) ⇒ Object

Send the data over the wire to this client.

data - The XML String or XML::Node to write to the socket.

Returns nothing.



141
142
143
144
145
146
147
 
# File 'lib/vines/stream.rb', line 141

def write(data)
  log_node(data, :out)
  if data.respond_to?(:to_xml)
    data = data.to_xml(:indent => 0)
  end
  send_data(data)
end