Class: OnStomp::Client

Inherits:

Object

• Object
• OnStomp::Client

show all

Includes:

OnStomp::Components::Scopes, Interfaces::ClientConfigurable, Interfaces::ClientEvents, Interfaces::FrameMethods, Interfaces::ReceiptManager, Interfaces::SubscriptionManager

Defined in:

lib/onstomp/client.rb,
lib/onstomp/failover/new_with_failover.rb

Overview

Instance Attribute Summary (collapse)

• - (OnStomp::Connections::Base) connection readonly

  Connection object specific to the established STOMP protocol version.

• - (Array<Fixnum>) heartbeats

  The client-side heartbeat settings to allow for this connection.

• - (String) host

  The host header value to send to the broker when connecting.

• - (String) login

  The login header value to send to the broker when connecting.

• - (String) passcode

  The passcode header value to send to the broker when connecting.

• - (Fixnum) read_timeout

  The number of seconds to wait before a connection that is read-blocked during the connect phase is considered dead.

• - (Symbol => Object) ssl readonly

  SSL options for the connection.

• - (String) uri readonly

  The URI reference to the STOMP broker.

• - (Array<String>) versions

  The protocol versions to allow for this connection.

• - (Fixnum) write_timeout

  The number of seconds to wait before a write-blocked connection is considered dead.

Methods you ought not use directly. (collapse)

• - (Object) dispatch_received(frame)

  Called by #connection when a frame has been read from the socket connection to the STOMP broker.

• - (Object) dispatch_transmitted(frame)

  Called by #connection when a frame has been written to the socket connection to the STOMP broker.

• - (OnStomp::Components::Frame) transmit(frame, cbs = {})

  Ultimately sends a frame to the STOMP broker.

Class Method Summary (collapse)

• + (OnStomp::Client, OnStomp::Failover::Client) new_with_failover(uri, options = {}) (also: new)

  Creates an alias chain for new so that if a failover: URI or an array of URIs are passed to the constructor, a failover client is built instead.

Instance Method Summary (collapse)

• - (self) close!

  Forces the connection between broker and client closed.

• - (self) connect(headers = {}) (also: #open)

  Connects to the STOMP broker referenced by #uri.

• - (true, false) connected?

  Returns true if a connection to the broker exists and itself is connected.

• - (OnStomp::Components::Frame) disconnect_with_flush(headers = {}) (also: #disconnect)

  Sends a DISCONNECT frame to the broker and blocks until the connection has been closed.

• - (Client) initialize(uri, options = {}) constructor

  Creates a new client for the specified uri and optional hash of options.

• - (Class) processor

  The class to use when instantiating a new IO processor for the connection.

Methods included from OnStomp::Components::Scopes

#transaction, #with_headers, #with_receipt

Methods included from Interfaces::SubscriptionManager

#subscriptions

Methods included from Interfaces::ClientEvents

#:begin, #abort, #ack, #after_receiving, #after_transmitting, #before_abort, #before_ack, #before_begin, #before_broker_beat, #before_client_beat, #before_commit, #before_disconnect, #before_error, #before_message, #before_nack, #before_receipt, #before_receiving, #before_send, #before_subscribe, #before_transmitting, #before_unsubscribe, #broker_beat, #client_beat, #commit, #error, #message, #nack, #on_abort, #on_ack, #on_begin, #on_broker_beat, #on_client_beat, #on_commit, #on_connect, #on_disconnect, #on_error, #on_message, #on_nack, #on_receipt, #on_send, #on_subscribe, #on_unsubscribe, #pending_connection_events, #receipt, #receiving, #send, #subscribe, #transmitting, #trigger_after_receiving, #trigger_after_transmitting, #trigger_before_receiving, #trigger_before_transmitting, #trigger_frame_event, #unsubscribe

Methods included from Interfaces::EventManager

#bind_event, #event_callbacks, included, #trigger_event

Methods included from Interfaces::FrameMethods

#abort, #ack, #beat, #begin, #commit, #nack, #send, #subscribe, #unsubscribe

Methods included from Interfaces::ClientConfigurable

included

65
66
67
68
69
70
71
72
73
74

# File 'lib/onstomp/client.rb', line 65

def initialize uri, options={}
  @uri = uri.is_a?(::URI) ? uri : ::URI.parse(uri)
  @ssl = options.delete(:ssl)
  configure_configurable options
  configure_subscription_management
  configure_receipt_management
  on_disconnect do |f, con|
    close unless f[:receipt]
  end
end

21
22
23

# File 'lib/onstomp/client.rb', line 21

def connection
  @connection
end

29

# File 'lib/onstomp/client.rb', line 29

attr_configurable_client_beats :heartbeats

35

# File 'lib/onstomp/client.rb', line 35

attr_configurable_str :host, :default => 'localhost', :uri_attr => :host

39

# File 'lib/onstomp/client.rb', line 39

attr_configurable_str :login, :default => '', :uri_attr => :user

43

# File 'lib/onstomp/client.rb', line 43

attr_configurable_str :passcode, :default => '', :uri_attr => :password

59

# File 'lib/onstomp/client.rb', line 59

attr_configurable_int :read_timeout, :default => 120

18
19
20

# File 'lib/onstomp/client.rb', line 18

def ssl
  @ssl
end

15
16
17

# File 'lib/onstomp/client.rb', line 15

def uri
  @uri
end

25

# File 'lib/onstomp/client.rb', line 25

attr_configurable_protocols :versions

53

# File 'lib/onstomp/client.rb', line 53

attr_configurable_int :write_timeout, :default => 120

9
10
11
12
13
14
15

# File 'lib/onstomp/failover/new_with_failover.rb', line 9

def new_with_failover(uri, options={})
  if uri.is_a?(Array) || uri.to_s =~ /^failover:/i
    OnStomp::Failover::Client.new(uri, options)
  else
    new_without_failover(uri, options)
  end
end

123
124
125
126
127

# File 'lib/onstomp/client.rb', line 123

def close!
  close
  processor_inst.stop
  self
end

81
82
83
84
85
86
87
88
89
90
91
92

# File 'lib/onstomp/client.rb', line 81

def connect headers={}
  # FIXME: This is a quick fix to force the Threaded IO processor to
  # complete its work before we establish a connection.
  processor_inst.stop
  @connection = OnStomp::Connections.connect self, headers,
    { :accept-version' => @versions.join(','), :host => @host,
      :heart-beat' => @heartbeats.join(','), :login => @login,
      :passcode => @passcode }, pending_connection_events,
    read_timeout, write_timeout
  processor_inst.start
  self
end

113
114
115

# File 'lib/onstomp/client.rb', line 113

def connected?
  connection && connection.connected?
end

101
102
103
104
105
106

# File 'lib/onstomp/client.rb', line 101

def disconnect_with_flush headers={}
  processor_inst.prepare_to_close
  disconnect_without_flush(headers).tap do
    processor_inst.join
  end
end

145
146
147
148

# File 'lib/onstomp/client.rb', line 145

def dispatch_received frame
  trigger_before_receiving frame
  trigger_after_receiving frame
end

152
153
154

# File 'lib/onstomp/client.rb', line 152

def dispatch_transmitted frame
  trigger_after_transmitting frame
end

48

# File 'lib/onstomp/client.rb', line 48

attr_configurable_processor :processor

135
136
137
138
139
140
141

# File 'lib/onstomp/client.rb', line 135

def transmit frame, cbs={}
  frame.tap do
    register_callbacks frame, cbs
    trigger_before_transmitting frame
    connection && connection.write_frame_nonblock(frame)
  end
end