Class: Async::Bus::Protocol::Transaction

Inherits:
Object
  • Object
show all
Defined in:
lib/async/bus/protocol/transaction.rb

Overview

Represents a transaction for a remote procedure call.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection, id, timeout: nil) ⇒ Transaction

Initialize a new transaction.



17
18
19
20
21
22
23
24
25
 
# File 'lib/async/bus/protocol/transaction.rb', line 17

def initialize(connection, id, timeout: nil)
  @connection = connection
  @id = id
  
  @timeout = timeout
  
  @received = Thread::Queue.new
  @accept = nil
end

Instance Attribute Details

#accept(object, arguments, options, block_given) ⇒ Object (readonly)

Accept a remote procedure invocation.



40
41
42
 
# File 'lib/async/bus/protocol/transaction.rb', line 40

def accept
  @accept
end

#connection ⇒ Object (readonly)

Returns the value of attribute connection.



28
29
30
 
# File 'lib/async/bus/protocol/transaction.rb', line 28

def connection
  @connection
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



31
32
33
 
# File 'lib/async/bus/protocol/transaction.rb', line 31

def id
  @id
end

#received ⇒ Object (readonly)

Returns the value of attribute received.



37
38
39
 
# File 'lib/async/bus/protocol/transaction.rb', line 37

def received
  @received
end

#The queue of received messages.(queueofreceivedmessages.) ⇒ Object (readonly) 



37
 
# File 'lib/async/bus/protocol/transaction.rb', line 37

attr :received

#The transaction ID.(transactionID.) ⇒ Object (readonly) 



31
 
# File 'lib/async/bus/protocol/transaction.rb', line 31

attr :id

#timeout ⇒ Object

Returns the value of attribute timeout.



34
35
36
 
# File 'lib/async/bus/protocol/transaction.rb', line 34

def timeout
  @timeout
end

Instance Method Details

#close ⇒ Object

Close the transaction and clean up resources.



73
74
75
76
77
78
79
80
 
# File 'lib/async/bus/protocol/transaction.rb', line 73

def close
  if connection = @connection
    @connection = nil
    @received.close
    
    connection.transactions.delete(@id)
  end
end

#invoke(name, arguments, options, &block) ⇒ Object

Invoke a remote procedure.



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/async/bus/protocol/transaction.rb', line 88

def invoke(name, arguments, options, &block)
  Console.debug(self){[name, arguments, options, block]}
  
  self.write(Invoke.new(@id, name, arguments, options, block_given?))
  
  while response = self.read
    case response
    when Return
      return response.result
    when Yield
      begin
        result = yield(*response.result)
        self.write(Next.new(@id, result))
      rescue => error
        self.write(Error.new(@id, error))
      end
    when Error
      raise(response.result)
    when Throw
      # Re-throw the tag and value that was thrown on the server side
      # Throw.result contains [tag, value] array
      tag, value = response.result
      throw(tag, value)
    end
  end
end

#push(message) ⇒ Object

Push a message to the transaction’s received queue. Silently ignores messages if the queue is already closed.



66
67
68
69
70
 
# File 'lib/async/bus/protocol/transaction.rb', line 66

def push(message)
  @received.push(message)
rescue ClosedQueueError
  # Queue is closed (transaction already finished/closed) - ignore silently.
end

#read ⇒ Object

Read a message from the transaction queue.



44
45
46
47
48
49
50
 
# File 'lib/async/bus/protocol/transaction.rb', line 44

def read
  if @received.empty?
    @connection.flush
  end
  
  @received.pop(timeout: @timeout)
end

#The accept handler.=(accepthandler. = (value)) ⇒ Object 



40
 
# File 'lib/async/bus/protocol/transaction.rb', line 40

attr :accept

#The connection for this transaction.=(connection) ⇒ Object 



28
 
# File 'lib/async/bus/protocol/transaction.rb', line 28

attr :connection

#The timeout for the transaction.=(timeout) ⇒ Object 



34
 
# File 'lib/async/bus/protocol/transaction.rb', line 34

attr_accessor :timeout

#write(message) ⇒ Object

Write a message to the connection.



55
56
57
58
59
60
61
 
# File 'lib/async/bus/protocol/transaction.rb', line 55

def write(message)
  if @connection
    @connection.write(message)
  else
    raise RuntimeError, "Transaction is closed!"
  end
end