Class: OpenAI::Realtime::Recovery Private

Inherits:
Object
  • Object
show all
Defined in:
lib/openai/helpers/realtime/recovery.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Owns successive block-scoped sockets on one Async reactor. Calls from other fibers or threads use mailboxes; the physical socket never leaves its owner. No incoming events are prefetched: a receive request drives each read.

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(manager:, max_reconnect_attempts:, max_queue_bytes:, on_reconnected:) ⇒ Recovery

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of Recovery.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 23

def initialize(manager:, max_reconnect_attempts:, max_queue_bytes:, on_reconnected:)
  @manager = manager
  @max_attempts = max_reconnect_attempts
  @max_queue_bytes = max_queue_bytes
  @on_reconnected = on_reconnected
  @mutex = Mutex.new
  @commands = Queue.new
  @reads = Queue.new
  @stop_requests = Queue.new
  @ready = Queue.new
  @waiters = {}
  @pending = []
  @pending_bytes = 0
  @state = :connecting
  @generation = 0
  @attempts = 0
  @flushing = false
end

Class Method Details

.validate_options!(max_reconnect_attempts:, max_queue_bytes:, on_reconnected:) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



11
12
13
14
15
16
17
18
19
20
21
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 11

def self.validate_options!(max_reconnect_attempts:, max_queue_bytes:, on_reconnected:)
  {max_reconnect_attempts: max_reconnect_attempts, max_queue_bytes: max_queue_bytes}.each do |name, value|
    unless value.is_a?(Integer) && value >= 0
      raise ArgumentError, "`#{name}` must be a nonnegative Integer."
    end
  end

  unless on_reconnected.nil? || on_reconnected.respond_to?(:call)
    raise ArgumentError, "`on_reconnected` must respond to `call`."
  end
end

Instance Method Details

#abort ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



147
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 147

def abort = stop(nil)

#close(code: 1000, reason: "") ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Cancel outstanding work and make a bounded attempt to send the close frame.



143
144
145
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 143

def close(code: 1000, reason: "")
  stop({code: code, reason: reason})
end

#closed? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:



69
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 69

def closed? = @mutex.synchronize { @state == :closed }

#flush_pending ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 125

def flush_pending
  if Fiber.current.equal?(@owner_fiber)
    raise state_error("Queue flush interrupted; remaining events were not sent.") unless flush
    return nil
  end

  reply = @mutex.synchronize do
    raise @error, cause: nil if @error
    raise state_error("Cannot flush until the replacement connection is ready.") unless @state == :open

    register(:flush).tap { @commands << [:flush, nil, _1, @generation] }
  end

  await_reply(reply)
  nil
end

#inspect ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



42
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 42

def inspect = "#<#{self.class.name}>"

#open ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 44

def open
  load_async
  ::Kernel.Sync() do |root|
    stopper = root.async do
      @stop_requests.pop
      @lifecycle&.stop
    end

    @lifecycle = root.async { run }
    begin
      kind, value = @ready.pop
      raise value, cause: nil if kind == :error

      result = yield(value)
      raise @error, cause: nil if @error

      result
    ensure
      $! ? abort : close
      @lifecycle.wait
      stopper.stop
    end
  end
end

#pending_messages ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



75
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 75

def pending_messages = @mutex.synchronize { @pending.dup }

#read ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 88

def read
  return @physical.receive_raw if Fiber.current.equal?(@owner_fiber)

  reply = @mutex.synchronize do
    raise @error, cause: nil if @error
    return nil if @state == :closed

    register(:read).tap { @reads << _1 }
  end

  await_reply(reply)
end

#reconnecting? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:



71
72
73
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 71

def reconnecting?
  @mutex.synchronize { @state == :recovering || @state == :restoring }
end

#take_pending_messages ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



77
78
79
80
81
82
83
84
85
86
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 77

def take_pending_messages
  @mutex.synchronize do
    raise state_error("Cannot remove queued events while flushing.") if @flushing

    messages = @pending
    @pending = []
    @pending_bytes = 0
    messages
  end
end

#write(text) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/openai/helpers/realtime/recovery.rb', line 101

def write(text)
  return write_once(text) if Fiber.current.equal?(@owner_fiber)

  reply = @mutex.synchronize do
    raise write_error(text), cause: nil if @error || [:closed, :ending].include?(@state)

    if @state != :open || @flushing
      if @max_queue_bytes.zero? || @pending_bytes + [text.bytesize, 1].max > @max_queue_bytes
        raise OpenAI::Errors::RealtimeQueueFullError
      end

      @pending << text.dup.freeze
      @pending_bytes += [text.bytesize, 1].max
      return nil
    end

    encoded = text.dup.freeze
    register(:write, encoded).tap { @commands << [:write, encoded, _1, @generation] }
  end

  await_reply(reply)
  nil
end