Class: Teams::HttpStream

Inherits:
Object
  • Object
show all
Defined in:
lib/teams/http_stream.rb

Overview

HTTP-based streaming for Microsoft Teams activities.

Emits are queued and flushed by a background thread, like the TypeScript and Python streamers: each flush drains the whole queue, coalescing the accumulated message text into one typing chunk, and consecutive flushes are spaced apart to stay under Teams rate limits. close waits for the queue to drain before sending the final message.

Constant Summary collapse

STREAM_CHANNEL_DATA_KEYS = 
%w[streamId streamType streamSequence].freeze
NON_RETRYABLE_ERRORS = 
[TerminalStreamError, StreamCancelledError].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(app:, conversation_reference:) ⇒ HttpStream

Returns a new instance of HttpStream.



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/teams/http_stream.rb', line 17

def initialize(app:, conversation_reference:)
  @app = app
  @conversation_reference = conversation_reference
  @mutex = Mutex.new
  @flusher = nil
  @flushing = false
  @flush_interval = 0.5
  @poll_interval = 0.1
  @total_wait_timeout = 30
  # Chunk sends use the Python streamer's tuned retry options; other
  # stream sends use the shared defaults.
  @chunk_retry = { max_attempts: 8, delay: 0.5, max_delay: 4.0, jitter: :none }
  @send_retry = { max_attempts: 5, delay: 0.5, max_delay: 30.0, jitter: :full }
  reset_state
  @result = nil
  @canceled = false
  @timed_out = false
  @chunk_handlers = []
  @close_handlers = []
end

Instance Attribute Details

#app ⇒ Object (readonly)

Returns the value of attribute app.



15
16
17
 
# File 'lib/teams/http_stream.rb', line 15

def app
  @app
end

#conversation_reference ⇒ Object (readonly)

Returns the value of attribute conversation_reference.



15
16
17
 
# File 'lib/teams/http_stream.rb', line 15

def conversation_reference
  @conversation_reference
end

Instance Method Details

#canceled ⇒ Object 



52
53
54
 
# File 'lib/teams/http_stream.rb', line 52

def canceled
  @canceled
end

#clear_text ⇒ Object 



97
98
99
100
101
102
103
 
# File 'lib/teams/http_stream.rb', line 97

def clear_text
  @mutex.synchronize do
    @text = +""
    @queue.reject! { |activity| activity["type"] == "message" }
    @final_activity = nil
  end
end

#close ⇒ Object 



105
106
107
108
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
 
# File 'lib/teams/http_stream.rb', line 105

def close
  return @result if closed
  return nil if canceled
  return nil if no_content_pending?

  return nil unless wait_for_flush

  return nil if canceled

  unless @id
    app.logger&.warn("no stream id set, cannot close stream")
    return nil
  end

  return nil unless final_content?

  # Merging the sent activity with the response keeps the id available
  # even though live Teams answers follow-up stream posts with 202 and an
  # empty body. @result doubling as the closed flag depends on this.
  @result = if @timed_out
    send_final
  else
    begin
      outbound = final_stream_activity
      Api::SentActivity.merge(outbound, send_with_retry(outbound, @send_retry))
    rescue StreamTimedOutError
      # The final streamed send tripped the two-minute limit. Update the
      # original message in place with the buffered content instead of
      # posting a duplicate.
      send_final
    end
  end

  @close_handlers.each { |handler| handler.call(@result) }
  @result
ensure
  @mutex.synchronize { reset_state } if @result
end

#closed ⇒ Object 



60
61
62
 
# File 'lib/teams/http_stream.rb', line 60

def closed
  !@result.nil?
end

#count ⇒ Object 



64
65
66
 
# File 'lib/teams/http_stream.rb', line 64

def count
  @mutex.synchronize { @queue.length }
end

#emit(activity_or_text) ⇒ Object

Raises:



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/teams/http_stream.rb', line 72

def emit(activity_or_text)
  raise StreamCancelledError, "Stream has been cancelled." if canceled

  activity = normalize_activity(activity_or_text)

  @mutex.synchronize do
    # Emitting after close reopens the stream: start a new streamed
    # message on the same instance. The canceled flag stays sticky.
    reset_for_next_stream if closed

    @queue << activity
    @flusher ||= Thread.new { run_flusher }
  end

  nil
end

#on_chunk(&handler) ⇒ Object

Registers a handler called with the SentActivity of every stream chunk. Handlers persist across stream reuse and run on the flusher thread.



40
41
42
43
 
# File 'lib/teams/http_stream.rb', line 40

def on_chunk(&handler)
  @chunk_handlers << handler
  self
end

#on_close(&handler) ⇒ Object

Registers a handler called with the final SentActivity when the stream closes. Handlers persist across stream reuse.



47
48
49
50
 
# File 'lib/teams/http_stream.rb', line 47

def on_close(&handler)
  @close_handlers << handler
  self
end

#sequence ⇒ Object 



68
69
70
 
# File 'lib/teams/http_stream.rb', line 68

def sequence
  @mutex.synchronize { @sequence }
end

#timed_out ⇒ Object 



56
57
58
 
# File 'lib/teams/http_stream.rb', line 56

def timed_out
  @timed_out
end

#update(text) ⇒ Object 



89
90
91
92
93
94
95
 
# File 'lib/teams/http_stream.rb', line 89

def update(text)
  emit(
    "type" => "typing",
    "text" => text,
    "channelData" => { "streamType" => "informative" }
  )
end