Angular客户端无法创建SignalR上传流，调用send方法报错

核心问题定位 #

你遇到的TypeError: streams[streamId].subscribe is not a function错误，大概率是SignalR流参数的类型不匹配或RxJS与SignalR的Subject混用导致的，结合你调试中发现streamId变为"distinct"的现象，更指向后者的干扰。

具体解决方案 #

1. 确保使用SignalR原生Subject #

首先确认你导入的是SignalR自带的Subject，而非RxJS的Subject：

// 正确导入
import { Subject } from '@microsoft/signalr';

// 错误示例（务必避免）
// import { Subject } from 'rxjs';

可通过运行时校验确认类型：

const subject: signalR.Subject<number[]> = new signalR.Subject<number[]>();
console.log(subject instanceof signalR.Subject); // 应输出true

2. 禁止对SignalR Subject使用RxJS操作符 #

如果你在传入send方法前，对Subject使用了RxJS的pipe、distinct等操作符，会返回一个Observable而非SignalR Subject，这会导致SignalR内部处理流时出错：

// 错误示例：
const stream$ = subject.pipe(distinct());
this.hubConn.send('UploadStream', 'someId', file.name, stream$); // 传入了Observable而非Subject

// 正确写法：直接传入原始SignalR Subject
this.hubConn.send('UploadStream', 'someId', file.name, subject);

3. 校验服务器端Hub方法签名 #

确保服务器端的UploadStream方法正确接收流参数，示例如下（.NET Core 7）：

public async Task UploadStream(string someId, string fileName, IAsyncEnumerable<byte[]> stream)
{
    // 处理文件块逻辑
    await foreach (var chunk in stream)
    {
        // 写入文件或存储
    }
}

4. 完善流的错误与完成处理 #

按照SignalR流式上传的规范，需手动管理流的分块发送、错误通知与完成信号：

const subject: signalR.Subject<number[]> = new signalR.Subject<number[]>();

// 订阅流的错误（捕获内部订阅问题）
subject.subscribe({
    error: (err) => console.error('Stream subscription error:', err)
});

// 初始化SignalR发送
this.hubConn.send('UploadStream', 'someId', file.name, subject)
    .catch(err => console.error('Send initiation error:', err));

// 文件分块读取与发送逻辑
const chunkSize = 1024 * 1024; // 1MB分块
let offset = 0;
const reader = new FileReader();

const readChunk = () => {
    const chunk = file.slice(offset, offset + chunkSize);
    reader.readAsArrayBuffer(chunk);
};

reader.onload = (e) => {
    const buffer = e.target?.result as ArrayBuffer;
    const byteArray = Array.from(new Uint8Array(buffer));
    subject.next(byteArray);
    
    offset += chunkSize;
    if (offset < file.size) {
        readChunk();
    } else {
        subject.complete(); // 发送流完成信号
    }
};

reader.onerror = (err) => {
    subject.error(err); // 发送流错误信号
};

readChunk();

5. 版本兼容性检查 #

确保客户端@microsoft/signalr@7.0.9与服务器端Microsoft.AspNetCore.SignalR版本完全一致（均为7.0.x），版本 mismatch 也可能引发此类兼容性问题。

内容的提问来源于stack exchange，提问作者The J