ASP.NET Core SignalR でストリーミングを使う
作成者: Brennan Conroy
ASP.NET Core SignalR は、クライアントからサーバーへ、およびサーバーからクライアントへのストリーミングをサポートしています。 これは、時間の経過と共にデータのフラグメントを受信するシナリオに便利です。 ストリーミング時には、すべてのデータが使用可能になるまで待つのではなく、各フラグメントが使用可能になるとすぐにクライアントまたはサーバーに送信されます。
サンプル コードを表示またはダウンロードします (ダウンロード方法)。
ストリーミング用のハブを設定する
IAsyncEnumerable<T>、ChannelReader<T>、Task<IAsyncEnumerable<T>>、または Task<ChannelReader<T>> が返されると、ハブ メソッドは自動的にストリーミング ハブ メソッドになります。
サーバーからクライアントへのストリーミング
ストリーミング ハブ メソッドからは、ChannelReader<T> に加えて IAsyncEnumerable<T> を返すことができます。 IAsyncEnumerable<T> を返す最もシンプルな方法は、次のサンプルに示すように、ハブ メソッドを非同期反復子メソッドにすることです。 ハブ非同期反復子メソッドは、クライアントがストリームからサブスクライブを解除したときにトリガーされる CancellationToken パラメーターを受け取ることができます。 非同期反復子メソッドを使うと、ChannelReader が適切な速さで返されない、ChannelWriter<T> を完了せずにメソッドが終了するなど、チャネルでよく発生する問題を回避できます。
Note
次の例では、C# 8.0 以降が必要です。
public class AsyncEnumerableHub : Hub
{
public async IAsyncEnumerable<int> Counter(
int count,
int delay,
[EnumeratorCancellation]
CancellationToken cancellationToken)
{
for (var i = 0; i < count; i++)
{
// Check the cancellation token regularly so that the server will stop
// producing items if the client disconnects.
cancellationToken.ThrowIfCancellationRequested();
yield return i;
// Use the cancellationToken in other APIs that accept cancellation
// tokens so the cancellation can flow down to them.
await Task.Delay(delay, cancellationToken);
}
}
}
次のサンプルは、チャネルを使ってクライアントにデータをストリーミングする基本的な方法を示しています。 ChannelWriter<T> にオブジェクトが書き込まれるたびに、そのオブジェクトは直ちにクライアントに送信されます。 最後に ChannelWriter が完了し、ストリームが閉じられたことがクライアントに通知されます。
Note
バックグラウンド スレッドで ChannelWriter<T> に書き込み、できるだけ早く ChannelReader を返します。 ChannelReader が返されるまで、他のハブの呼び出しはブロックされます。
ロジックを try ... catch ステートメントでラップします。 finally ブロックで Channel を完了します。 エラーをフローしたい場合は、catch ブロック内でキャプチャし、finally ブロック内で書き込みます。
public ChannelReader<int> Counter(
int count,
int delay,
CancellationToken cancellationToken)
{
var channel = Channel.CreateUnbounded<int>();
// We don't want to await WriteItemsAsync, otherwise we'd end up waiting
// for all the items to be written before returning the channel back to
// the client.
_ = WriteItemsAsync(channel.Writer, count, delay, cancellationToken);
return channel.Reader;
}
private async Task WriteItemsAsync(
ChannelWriter<int> writer,
int count,
int delay,
CancellationToken cancellationToken)
{
Exception localException = null;
try
{
for (var i = 0; i < count; i++)
{
await writer.WriteAsync(i, cancellationToken);
// Use the cancellationToken in other APIs that accept cancellation
// tokens so the cancellation can flow down to them.
await Task.Delay(delay, cancellationToken);
}
}
catch (Exception ex)
{
localException = ex;
}
finally
{
writer.Complete(localException);
}
}
サーバーからクライアントへのストリーミング ハブ メソッドは、クライアントがストリームからサブスクライブを解除したときにトリガーされる CancellationToken パラメーターを受け取ることができます。 クライアントがストリームの終了前に切断した場合は、このトークンを使ってサーバーの操作を停止し、リソースを解放します。
クライアントからサーバーへのストリーミング
型 ChannelReader<T> または IAsyncEnumerable<T> の 1 つ以上のオブジェクトを受け取ると、ハブ メソッドは自動的にクライアントからサーバーへのストリーミング ハブ メソッドになります。 次のサンプルは、クライアントから送信されたストリーミング データを読み取る基本を示しています。 クライアントから ChannelWriter<T> に書き込まれるたびに、ハブ メソッドから読み取られるサーバーの ChannelReader にデータが書き込まれます。
public async Task UploadStream(ChannelReader<string> stream)
{
while (await stream.WaitToReadAsync())
{
while (stream.TryRead(out var item))
{
// do something with the stream item
Console.WriteLine(item);
}
}
}
IAsyncEnumerable<T> バージョンのメソッドが続きます。
Note
次の例では、C# 8.0 以降が必要です。
public async Task UploadStream(IAsyncEnumerable<string> stream)
{
await foreach (var item in stream)
{
Console.WriteLine(item);
}
}
.NET クライアント
サーバーからクライアントへのストリーミング
HubConnection の StreamAsync と StreamAsChannelAsync のメソッドは、サーバーからクライアントへのストリーミング メソッドを呼び出すために使われます。 ハブ メソッド名と、ハブ メソッドに定義されている引数を StreamAsync または StreamAsChannelAsync に渡します。 StreamAsync<T> と StreamAsChannelAsync<T> のジェネリック パラメーターには、ストリーミング メソッドから返されるオブジェクトの型を指定します。 型 IAsyncEnumerable<T> または ChannelReader<T> のオブジェクトは、ストリームの呼び出しから返され、クライアント上のストリームを表します。
IAsyncEnumerable<int> を返す StreamAsync の例:
// Call "Cancel" on this CancellationTokenSource to send a cancellation message to
// the server, which will trigger the corresponding token in the hub method.
var cancellationTokenSource = new CancellationTokenSource();
var stream = hubConnection.StreamAsync<int>(
"Counter", 10, 500, cancellationTokenSource.Token);
await foreach (var count in stream)
{
Console.WriteLine($"{count}");
}
Console.WriteLine("Streaming completed");
ChannelReader<int> を返す、対応する StreamAsChannelAsync の例:
// Call "Cancel" on this CancellationTokenSource to send a cancellation message to
// the server, which will trigger the corresponding token in the hub method.
var cancellationTokenSource = new CancellationTokenSource();
var channel = await hubConnection.StreamAsChannelAsync<int>(
"Counter", 10, 500, cancellationTokenSource.Token);
// Wait asynchronously for data to become available
while (await channel.WaitToReadAsync())
{
// Read all currently available data synchronously, before waiting for more data
while (channel.TryRead(out var count))
{
Console.WriteLine($"{count}");
}
}
Console.WriteLine("Streaming completed");
上のコードでは、次のようになります。
HubConnectionに対するStreamAsChannelAsyncメソッドは、サーバーからクライアントへのストリーミング メソッドを呼び出すために使われます。 ハブ メソッド名と、ハブ メソッドに定義されている引数をStreamAsChannelAsyncに渡します。StreamAsChannelAsync<T>のジェネリック パラメーターには、ストリーミング メソッドから返されるオブジェクトの型を指定します。ChannelReader<T>は、ストリームの呼び出しから返され、クライアント上のストリームを表します。
クライアントからサーバーへのストリーミング
クライアントからサーバーへのストリーミング ハブ メソッドを .NET クライアントから呼び出す方法は 2 つあります。 呼び出されたハブ メソッドに応じて、SendAsync、InvokeAsync、StreamAsChannelAsync の引数として IAsyncEnumerable<T> または ChannelReader を渡すことができます。
IAsyncEnumerable または ChannelWriter オブジェクトにデータが書き込まれるたびに、サーバー上のハブ メソッドは、クライアントからのデータを含む新しい項目を受け取ります。
IAsyncEnumerable オブジェクトを使っている場合、ストリーム項目を返すメソッドが終了すると、ストリームは終了します。
Note
次の例では、C# 8.0 以降が必要です。
async IAsyncEnumerable<string> clientStreamData()
{
for (var i = 0; i < 5; i++)
{
var data = await FetchSomeData();
yield return data;
}
//After the for loop has completed and the local function exits the stream completion will be sent.
}
await connection.SendAsync("UploadStream", clientStreamData());
また、ChannelWriter を使っている場合は、channel.Writer.Complete() を使ってチャネルを完了します。
var channel = Channel.CreateBounded<string>(10);
await connection.SendAsync("UploadStream", channel.Reader);
await channel.Writer.WriteAsync("some data");
await channel.Writer.WriteAsync("some more data");
channel.Writer.Complete();
JavaScript クライアント
サーバーからクライアントへのストリーミング
JavaScript クライアントの場合、connection.stream を使ってハブ上でサーバーからクライアントへのストリーミング メソッドを呼び出します。 stream メソッドは、2 つの引数を受け取ります。
- ハブ メソッドの名前。 次の例では、ハブ メソッドの名前は
Counterです。 - ハブ メソッドで定義されている引数。 次の例の引数は、受信するストリーム項目の数とストリーム項目間の延期期間のカウントです。
connection.stream から IStreamResult が返されます。これには subscribe メソッドが含まれています。 subscribe に IStreamSubscriber を渡し、stream の呼び出しからの通知を受け取る next、error、complete のコールバックを設定します。
connection.stream("Counter", 10, 500)
.subscribe({
next: (item) => {
var li = document.createElement("li");
li.textContent = item;
document.getElementById("messagesList").appendChild(li);
},
complete: () => {
var li = document.createElement("li");
li.textContent = "Stream completed";
document.getElementById("messagesList").appendChild(li);
},
error: (err) => {
var li = document.createElement("li");
li.textContent = err;
document.getElementById("messagesList").appendChild(li);
},
});
クライアントからストリームを終了するには、subscribe メソッドから返された ISubscription に対して dispose メソッドを呼び出します。 このメソッドを呼び出すと、ハブ メソッドの CancellationToken パラメーターを指定していた場合にそれが取り消されます。
クライアントからサーバーへのストリーミング
JavaScript クライアントからは、呼び出されたハブ メソッドに応じて send、invoke、または stream の引数として Subject を渡すことで、ハブに対するクライアントからサーバーへのストリーミング メソッドが呼び出されます。 Subject は、Subject のようなクラスです。 たとえば、RxJS では、そのライブラリの Subject クラスを使うことができます。
const subject = new signalR.Subject();
yield connection.send("UploadStream", subject);
var iteration = 0;
const intervalHandle = setInterval(() => {
iteration++;
subject.next(iteration.toString());
if (iteration === 10) {
clearInterval(intervalHandle);
subject.complete();
}
}, 500);
項目を指定して subject.next(item) を呼び出すと、ストリームに項目が書き込まれ、ハブ メソッドはサーバー上の項目を受け取ります。
ストリームを終了するには、subject.complete() を呼び出します。
Java クライアント
サーバーからクライアントへのストリーミング
SignalR Java クライアントには、ストリーミング メソッドを呼び出すために stream メソッドが使われます。 stream は 3 つ以上の引数を受け取ります。
- ストリーム項目の想定される型。
- ハブ メソッドの名前。
- ハブ メソッドで定義されている引数。
hubConnection.stream(String.class, "ExampleStreamingHubMethod", "Arg1")
.subscribe(
(item) -> {/* Define your onNext handler here. */ },
(error) -> {/* Define your onError handler here. */},
() -> {/* Define your onCompleted handler here. */});
HubConnection に対する stream メソッドからは、ストリーム項目型の Observable が返されます。 Observable 型の subscribe メソッドには、onNext、onError、onCompleted のハンドラーが定義されています。
クライアントからサーバーへのストリーミング
SignalR Java クライアントからは、呼び出されたハブ メソッドに応じて send、invoke、または stream の引数として Observable を渡すことで、ハブに対するクライアントからサーバーへのストリーミング メソッドが呼び出されます。
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.send("UploadStream", stream);
stream.onNext("FirstItem");
stream.onNext("SecondItem");
stream.onComplete();
項目を指定して stream.onNext(item) を呼び出すと、ストリームに項目が書き込まれ、ハブ メソッドはサーバー上の項目を受け取ります。
ストリームを終了するには、stream.onComplete() を呼び出します。
その他のリソース
ASP.NET Core
フィードバック
以下は間もなく提供いたします。2024 年を通じて、コンテンツのフィードバック メカニズムとして GitHub の issue を段階的に廃止し、新しいフィードバック システムに置き換えます。 詳細については、「https://aka.ms/ContentUserFeedback」を参照してください。
フィードバックの送信と表示