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
v2.0.7

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

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