GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/teams.net

Public

mirrored from https://github.com/microsoft/teams.netAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
next/core-compat-sso-middleware

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

teams.net/core/src/Microsoft.Teams.Bot.Apps

core/src/Microsoft.Teams.Bot.Apps/TeamsStreamingWriter.cs

175lines · modecode

RawDownload
Latest commit unavailable.
1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4using Microsoft.Teams.Bot.Apps.Schema;
5using Microsoft.Teams.Bot.Apps.Schema.Entities;
6using Microsoft.Teams.Bot.Core;
7
8namespace Microsoft.Teams.Bot.Apps;
9
10/// <summary>
11/// Manages the send loop for Teams streaming messages.
12/// Callers append raw deltas; the writer accumulates them and sends the full
13/// text so far on each update. Every chunk — informative, intermediate, and
14/// final — is sent as a new POST with a shared <c>streamId</c>
15/// so Teams renders them as a single progressively-updating bubble.
16/// </summary>
17/// <remarks>
18/// Typical usage:
19/// <code>
20/// var writer = TeamsStreamingWriter.CreateFromContext(context);
21/// await writer.SendInformativeUpdateAsync("Thinking…"); //optional placeholder while the bot thinks
22/// await writer.AppendResponseAsync(" Hello");
23/// await writer.AppendResponseAsync(", world");
24/// await writer.FinalizeResponseAsync(); // sends accumulated " Hello, world"
25/// </code>
26///
27/// Entities and Attachments are only sent with the final message activity.
28/// Pass them directly to <see cref="FinalizeResponseAsync"/>:
29/// <code>
30/// await writer.FinalizeResponseAsync(
31/// entities: [new CitationEntity(...)],
32/// attachments: [new TeamsAttachment(...)]);
33/// </code>
34/// </remarks>
35public sealed class TeamsStreamingWriter
36{
37 // Teams streaming API enforces a rate limit; send intermediate updates at most once per interval.
38 private static readonly TimeSpan _minChunkInterval = TimeSpan.FromMilliseconds(500);
39
40 private readonly ConversationClient _client;
41 private readonly TeamsActivity _reference;
42 // Assigned from the server's 201 response after the first send; null until then.
43 private string? _streamId;
44 private int _sequence;
45 private bool _finalized;
46 private bool _cancelled;
47 private string _accumulated = string.Empty;
48 private DateTime _lastChunkSent = DateTime.MinValue;
49
50 internal TeamsStreamingWriter(ConversationClient client, TeamsActivity reference)
51 {
52 _client = client;
53 _reference = reference;
54 }
55
56 /// <summary>
57 /// Creates a <see cref="TeamsStreamingWriter"/> bound to the given context.
58 /// </summary>
59 public static TeamsStreamingWriter CreateFromContext<TActivity>(Context<TActivity> context) where TActivity : TeamsActivity
60 {
61 ArgumentNullException.ThrowIfNull(context);
62 return new TeamsStreamingWriter(context.TeamsBotApplication.ConversationClient, context.Activity);
63 }
64
65 /// <summary>
66 /// Sends an informative placeholder (streamType = "informative").
67 /// Optional — if omitted the first <see cref="AppendResponseAsync"/> call begins the stream.
68 /// </summary>
69 public async Task SendInformativeUpdateAsync(string text, CancellationToken cancellationToken = default)
70 {
71 if (_lastChunkSent > DateTime.MinValue)
72 throw new InvalidOperationException("Cannot send an informative update after streaming has started.");
73
74 _sequence++;
75 SendActivityResponse? response = await _client.SendActivityAsync(BuildActivity(text, StreamType.Informative), cancellationToken: cancellationToken).ConfigureAwait(false);
76 _streamId ??= response?.Id;
77 }
78
79 /// <summary>
80 /// Appends <paramref name="chunk"/> to the accumulated text and sends the
81 /// full accumulated text as an intermediate streaming update (streamType = "streaming").
82 /// </summary>
83 /// <exception cref="InvalidOperationException">Thrown if <see cref="FinalizeResponseAsync"/> has already been called.</exception>
84 public async Task AppendResponseAsync(string chunk, CancellationToken cancellationToken = default)
85 {
86 if (_finalized)
87 throw new InvalidOperationException("Cannot append after FinalizeResponseAsync has been called.");
88
89 if (_cancelled)
90 return;
91
92 _accumulated += chunk;
93
94 if (DateTime.UtcNow - _lastChunkSent < _minChunkInterval)
95 return;
96
97 _sequence++;
98 try
99 {
100 SendActivityResponse? response = await _client.SendActivityAsync(BuildActivity(_accumulated, StreamType.Streaming), cancellationToken: cancellationToken).ConfigureAwait(false);
101 _streamId ??= response?.Id;
102 _lastChunkSent = DateTime.UtcNow;
103 }
104 catch (HttpRequestException ex) when (ex.Message.Contains("Content stream was cancelled by user", StringComparison.OrdinalIgnoreCase))
105 {
106 _cancelled = true;
107 }
108 }
109
110 /// <summary>
111 /// Sends the accumulated text as the final update (streamType = "final") and marks the stream complete.
112 /// </summary>
113 /// <param name="attachments">Optional attachments to include in the final message activity.</param>
114 /// <param name="entities">Optional entities (e.g. citations, mentions) to include in the final message activity.</param>
115 /// <param name="feedbackEnabled">Whether to enable the feedback loop (thumbs up/down) on the final message.</param>
116 /// <param name="cancellationToken">Cancellation token.</param>
117 /// <exception cref="InvalidOperationException">Thrown if <see cref="FinalizeResponseAsync"/> has already been called, or if no content has been accumulated via <see cref="AppendResponseAsync"/>.</exception>
118 public async Task FinalizeResponseAsync(IList<TeamsAttachment>? attachments = null, IList<Entity>? entities = null, bool feedbackEnabled = false, CancellationToken cancellationToken = default)
119 {
120 if (_finalized)
121 throw new InvalidOperationException("Cannot finalize after FinalizeResponseAsync has already been called.");
122
123 if (_cancelled)
124 return;
125
126 if (string.IsNullOrEmpty(_accumulated) && (attachments == null || attachments.Count == 0))
127 throw new InvalidOperationException("Cannot finalize with no content. Call AppendResponseAsync at least once before FinalizeResponseAsync.");
128
129 await _client.SendActivityAsync(BuildActivity(_accumulated, StreamType.Final, attachments, entities, feedbackEnabled), cancellationToken: cancellationToken).ConfigureAwait(false);
130
131 _finalized = true;
132 }
133
134 private TeamsActivity BuildActivity(string text, string streamType, IList<TeamsAttachment>? attachments = null, IList<Entity>? entities = null, bool feedbackEnabled = false)
135 {
136 bool isFinal = streamType == StreamType.Final;
137
138 TeamsActivityBuilder builder;
139
140 if (isFinal)
141 {
142 StreamInfoEntity streamInfo = new() { StreamType = streamType };
143 if (_streamId != null)
144 streamInfo.StreamId = _streamId;
145
146 builder = new TeamsActivityBuilder(new MessageActivity(text))
147 .WithConversationReference(_reference)
148 .AddEntity(streamInfo);
149
150 if (entities != null)
151 foreach (Entity entity in entities)
152 builder.AddEntity(entity);
153
154 if (attachments?.Count > 0)
155 builder.WithAttachments(attachments);
156
157 TeamsActivity activity = builder.Build();
158 if (feedbackEnabled) activity.AddFeedback();
159 return activity;
160 }
161 else
162 {
163 StreamingActivity streaming = new(text);
164 streaming.StreamInfo.StreamType = streamType;
165 streaming.StreamInfo.StreamSequence = _sequence;
166 if (_streamId != null)
167 streaming.StreamInfo.StreamId = _streamId;
168
169 builder = new TeamsActivityBuilder(streaming)
170 .WithConversationReference(_reference);
171 }
172
173 return builder.Build();
174 }
175}
176