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
feature/pabot-httpcontext-botid

Branches

Tags

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

Clone

HTTPS

Download ZIP

teams.net/Libraries/Microsoft.Teams.Plugins/Microsoft.Teams.Plugins.AspNetCore

Libraries/Microsoft.Teams.Plugins/Microsoft.Teams.Plugins.AspNetCore/AspNetCorePlugin.Stream.cs

239lines · modecode

RawDownload
Latest commit unavailable.
1// Copyright (c) Microsoft Corporation. All rights reserved.
2// Licensed under the MIT License.
3
4using System.Collections.Concurrent;
5
6using Microsoft.Teams.Api;
7using Microsoft.Teams.Api.Activities;
8using Microsoft.Teams.Api.Entities;
9using Microsoft.Teams.Apps.Plugins;
10using Microsoft.Teams.Common.Logging;
11
12using static Microsoft.Teams.Common.Extensions.TaskExtensions;
13
14namespace Microsoft.Teams.Plugins.AspNetCore;
15
16public partial class AspNetCorePlugin
17{
18 public class Stream : IStreamer
19 {
20 public bool Closed => _closedAt is not null;
21 public int Count => _count;
22 public int Sequence => _index;
23
24 public required Func<IActivity, Task<IActivity>> Send { get; set; }
25 public ILogger? Logger { get; set; }
26 public event IStreamer.OnChunkHandler OnChunk = (_) => { };
27
28 protected int _index = 1;
29 protected string? _id;
30 protected string _text = string.Empty;
31 protected ChannelData _channelData = new();
32 protected List<Attachment> _attachments = [];
33 protected List<IEntity> _entities = [];
34 protected ConcurrentQueue<IActivity> _queue = [];
35
36 private DateTime? _closedAt;
37 private int _count = 0;
38 private MessageActivity? _result;
39 private readonly SemaphoreSlim _lock = new(1, 1);
40 private Timer? _timeout;
41
42 public void Emit(MessageActivity activity)
43 {
44 if (_timeout != null)
45 {
46 _timeout.Dispose();
47 _timeout = null;
48 }
49
50 _queue.Enqueue(activity);
51 _timeout = new Timer(_ =>
52 {
53 _ = FlushSafe();
54 }, null, 500, Timeout.Infinite);
55 }
56
57 public void Emit(TypingActivity activity)
58 {
59 if (_timeout != null)
60 {
61 _timeout.Dispose();
62 _timeout = null;
63 }
64
65 _queue.Enqueue(activity);
66 _timeout = new Timer(_ =>
67 {
68 _ = FlushSafe();
69 }, null, 500, Timeout.Infinite);
70 }
71
72 public void Emit(string text)
73 {
74 Emit(new MessageActivity(text));
75 }
76
77 public void Update(string text)
78 {
79 Emit(new TypingActivity(text)
80 {
81 ChannelData = new()
82 {
83 StreamType = StreamType.Informative
84 }
85 });
86 }
87
88 public async Task<MessageActivity?> Close(CancellationToken cancellationToken = default)
89 {
90 if (_index == 1 && _queue.Count == 0 && _lock.CurrentCount > 0) return null;
91 if (_result is not null) return _result;
92 // _lock.CurrentCount == 0 means Flush() is mid-await (queue drained but SendActivity calls
93 // still pending). Wait it out so the final message doesn't race in-flight chunks.
94 while (_id is null || _queue.Count > 0 || _lock.CurrentCount == 0)
95 {
96 await Task.Delay(50, cancellationToken).ConfigureAwait(false);
97 }
98
99 if (_text == string.Empty && _attachments.Count == 0) // when only informative updates are present
100 {
101 _text = "Streaming closed with no content";
102 }
103
104 var activity = new MessageActivity(_text)
105 .AddAttachment(_attachments.ToArray());
106
107 activity.WithId(_id);
108 activity.WithData(_channelData);
109 activity.AddEntity(_entities.ToArray());
110 activity.AddStreamFinal();
111
112 var res = await Retry(() => Send(activity)).ConfigureAwait(false);
113 OnChunk(res);
114
115 _result = activity;
116 _closedAt = DateTime.Now;
117 _index = 1;
118 _id = null;
119 _text = string.Empty;
120 _attachments = [];
121 _entities = [];
122 _channelData = new();
123
124 return (MessageActivity)res;
125 }
126
127 protected async Task Flush()
128 {
129 bool hasPendingState = _id is null && _count > 0 && _text != string.Empty;
130 if (_queue.Count == 0 && !hasPendingState) return;
131
132 await _lock.WaitAsync().ConfigureAwait(false);
133
134 try
135 {
136 if (_timeout != null)
137 {
138 _timeout.Dispose();
139 _timeout = null;
140 }
141
142 Queue<TypingActivity> informativeUpdates = new();
143 var dequeued = 0;
144
145 while (_queue.TryDequeue(out var activity))
146 {
147 if (activity is MessageActivity message)
148 {
149 _text += message.Text;
150 _attachments.AddRange(message.Attachments ?? []);
151 _entities.AddRange(message.Entities ?? []);
152 }
153
154 if (activity.ChannelData is not null)
155 {
156 _channelData = _channelData.Merge(activity.ChannelData);
157 }
158
159 if (activity is TypingActivity typing && typing.ChannelData?.StreamType == StreamType.Informative && _text == string.Empty)
160 {
161 // If `_text` is not empty then it's possible that streaming has started.
162 // And so informative updates cannot be sent.
163 informativeUpdates.Enqueue(typing);
164 }
165
166 dequeued++;
167 _count++;
168 }
169
170 // Recalculate inside the lock to account for any concurrent changes.
171 hasPendingState = _id is null && _count > 0 && _text != string.Empty;
172 if (dequeued == 0 && !hasPendingState) return;
173
174 // Send informative updates
175 if (informativeUpdates.Count > 0)
176 {
177 while (informativeUpdates.TryDequeue(out var typing))
178 {
179 await SendActivity(typing).ConfigureAwait(false);
180 }
181 }
182
183 // Send text chunk
184 if (_text != string.Empty)
185 {
186 var toSend = new TypingActivity(_text);
187 await SendActivity(toSend).ConfigureAwait(false);
188 }
189
190 if (_queue.Count > 0)
191 {
192 _timeout = new Timer(_ =>
193 {
194 _ = FlushSafe();
195 }, null, 500, Timeout.Infinite);
196 }
197
198 async Task SendActivity(TypingActivity toSend)
199 {
200 if (_id is not null)
201 {
202 toSend.WithId(_id);
203 }
204
205 toSend.AddStreamUpdate(_index);
206 var res = await Retry(() => Send(toSend)).ConfigureAwait(false);
207 OnChunk(res);
208 _id ??= res.Id;
209 _index++;
210 }
211 }
212 finally
213 {
214 _lock.Release();
215 }
216 }
217
218 private async Task FlushSafe()
219 {
220 try
221 {
222 await Flush().ConfigureAwait(false);
223 }
224 catch (Exception ex)
225 {
226 Logger?.Warn("Stream flush failed; will retry if there is pending state.", ex);
227 // Reschedule a retry so Close() doesn't spin forever
228 // waiting for _id to be set after a transient send failure.
229 if (_queue.Count > 0 || _id is null && _count > 0)
230 {
231 _timeout = new Timer(_ =>
232 {
233 _ = FlushSafe();
234 }, null, 1000, Timeout.Infinite);
235 }
236 }
237 }
238 }
239}