Class: UnionStationHooks::Context

Inherits:
Object
  • Object
show all
Includes:
Utils
Defined in:
lib/union_station_hooks_core/context.rb

Overview

A Context object is the “heart” of all ‘union_station_hooks_*` gems. It contains a connection to the UstRouter (through a Connection object) and allows you to create Transaction objects.

Context is a singleton. During initialization (‘UnionStationHooks.initialize!`), an instance is created and stored in `UnionStationHooks.context`. All the public API methods make use of this singleton context.

See hacking/Architecture.md for an overview.

Constant Summary collapse

RETRY_SLEEP = 
0.2
NETWORK_ERRORS = 
[
  Errno::EPIPE, Errno::ECONNREFUSED, Errno::ECONNRESET,
  Errno::EHOSTUNREACH, Errno::ENETDOWN, Errno::ENETUNREACH,
  Errno::ETIMEDOUT
]

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Utils

#base64, #connect_to_server, #encoded_timestamp, #get_socket_address_type, included, #local_socket_address?, #process_times, #process_ust_router_reply, #process_ust_router_reply_message, #require_key, #require_non_empty_key

Constructor Details

#initialize(ust_router_address, username, password, node_name) ⇒ Context

Returns a new instance of Context.



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/union_station_hooks_core/context.rb', line 58

def initialize(ust_router_address, username, password, node_name)
  @server_address = ust_router_address
  @username = username
  @password = password
  if node_name && !node_name.empty?
    @node_name = node_name
  else
    @node_name = `hostname`.strip
  end
  @random_dev = File.open('/dev/urandom')

  # This mutex protects the following instance variables, but
  # not the contents of @connection.
  @mutex = Mutex.new

  @connection = Connection.new(nil)
  if @server_address && local_socket_address?(@server_address)
    @max_connect_tries = 10
  else
    @max_connect_tries = 1
  end
  @reconnect_timeout = 1
  @next_reconnect_time = Time.utc(1980, 1, 1)
end

Instance Attribute Details

#max_connect_tries ⇒ Object

Returns the value of attribute max_connect_tries.



55
56
57
 
# File 'lib/union_station_hooks_core/context.rb', line 55

def max_connect_tries
  @max_connect_tries
end

#reconnect_timeout ⇒ Object

Returns the value of attribute reconnect_timeout.



56
57
58
 
# File 'lib/union_station_hooks_core/context.rb', line 56

def reconnect_timeout
  @reconnect_timeout
end

Instance Method Details

#clear_connection ⇒ Object 



89
90
91
92
93
94
95
96
97
 
# File 'lib/union_station_hooks_core/context.rb', line 89

def clear_connection
  @mutex.synchronize do
    @connection.synchronize do
      @random_dev = File.open('/dev/urandom') if @random_dev.closed?
      @connection.unref
      @connection = Connection.new(nil)
    end
  end
end

#close ⇒ Object 



99
100
101
102
103
104
105
106
107
 
# File 'lib/union_station_hooks_core/context.rb', line 99

def close
  @mutex.synchronize do
    @connection.synchronize do
      @random_dev.close
      @connection.unref
      @connection = nil
    end
  end
end

#connection ⇒ Object 



83
84
85
86
87
 
# File 'lib/union_station_hooks_core/context.rb', line 83

def connection
  @mutex.synchronize do
    @connection
  end
end

#continue_transaction(txn_id, group_name, category, key) ⇒ Object 



182
183
184
185
186
187
188
189
190
191
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
231
232
233
 
# File 'lib/union_station_hooks_core/context.rb', line 182

def continue_transaction(txn_id, group_name, category, key)
  if !@server_address
    return Transaction.new(nil, nil)
  elsif !txn_id || txn_id.empty?
    raise ArgumentError, 'Transaction ID may not be empty'
  end

  Lock.new(@mutex).synchronize do |_lock|
    if Time.now < @next_reconnect_time
      return Transaction.new(nil, nil)
    end

    Lock.new(@connection.mutex).synchronize do |connection_lock|
      if !@connection.connected?
        begin
          connect
          connection_lock.reset(@connection.mutex)
        rescue SystemCallError, IOError
          @connection.disconnect
          UnionStationHooks::Log.warn(
            "Cannot connect to the UstRouter at #{@server_address}; " \
            "retrying in #{@reconnect_timeout} second(s).")
          @next_reconnect_time = Time.now + @reconnect_timeout
          return Transaction.new(nil, nil)
        rescue Exception => e
          @connection.disconnect
          raise e
        end
      end

      begin
        @connection.channel.write('openTransaction',
          txn_id, group_name, '', category,
          Utils.encoded_timestamp,
          key,
          true)
        return Transaction.new(@connection, txn_id)
      rescue SystemCallError, IOError
        @connection.disconnect
        UnionStationHooks::Log.warn(
          "The UstRouter at #{@server_address}" \
          ' closed the connection; will reconnect in ' \
          "#{@reconnect_timeout} second(s).")
        @next_reconnect_time = Time.now + @reconnect_timeout
        return Transaction.new(nil, nil)
      rescue Exception => e
        @connection.disconnect
        raise e
      end
    end
  end
end

#new_transaction(group_name, category, key) ⇒ Object 



109
110
111
112
113
114
115
116
117
118
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
149
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
179
180
 
# File 'lib/union_station_hooks_core/context.rb', line 109

def new_transaction(group_name, category, key)
  if !@server_address
    return Transaction.new(nil, nil)
  elsif !group_name || group_name.empty?
    raise ArgumentError, 'Group name may not be empty'
  end

  txn_id = create_txn_id

  Lock.new(@mutex).synchronize do |_lock|
    if Time.now < @next_reconnect_time
      return Transaction.new(nil, nil)
    end

    Lock.new(@connection.mutex).synchronize do |connection_lock|
      if !@connection.connected?
        begin
          connect
          connection_lock.reset(@connection.mutex)
        rescue SystemCallError, IOError
          @connection.disconnect
          UnionStationHooks::Log.warn(
            "Cannot connect to the UstRouter at #{@server_address}; " \
            "retrying in #{@reconnect_timeout} second(s).")
          @next_reconnect_time = Time.now + @reconnect_timeout
          return Transaction.new(nil, nil)
        rescue Exception => e
          @connection.disconnect
          raise e
        end
      end

      begin
        @connection.channel.write('openTransaction',
          txn_id, group_name, '', category,
          Utils.encoded_timestamp,
          key,
          true,
          true)
        result = @connection.channel.read
        if result[0] != 'status'
          raise "Expected UstRouter to respond with 'status', " \
            "but got #{result.inspect} instead"
        elsif result[1] == 'ok'
          # Do nothing
        elsif result[1] == 'error'
          if result[2]
            raise "Unable to close transaction: #{result[2]}"
          else
            raise 'Unable to close transaction (no server message given)'
          end
        else
          raise "Expected UstRouter to respond with 'ok' or 'error', " \
            "but got #{result.inspect} instead"
        end

        return Transaction.new(@connection, txn_id)
      rescue SystemCallError, IOError
        @connection.disconnect
        UnionStationHooks::Log.warn(
          "The UstRouter at #{@server_address}" \
          ' closed the connection; will reconnect in ' \
          "#{@reconnect_timeout} second(s).")
        @next_reconnect_time = Time.now + @reconnect_timeout
        return Transaction.new(nil, nil)
      rescue Exception => e
        @connection.disconnect
        raise e
      end
    end
  end
end