serverpod-streams

Serverpod 流

接受或返回 Stream<T> 的端点方法会获得由 WebSocket 管理的客户端存根。类型必须可序列化。

将流与服务器事件结合使用是常见用法。参见 服务器事件。

定义流式方法

class ExampleEndpoint extends Endpoint {
  Stream<String> echoStream(Session session, Stream<String> stream) async* {
    await for (var message in stream) {
      yield message;
    }
  }
}

运行 serverpod generate 生成客户端存根。

客户端用法

var inStream = StreamController<String>();
var outStream = client.example.echoStream(inStream.stream);

outStream.listen((message) => print('Received: $message'));
inStream.add('Hello');

完成后关闭 StreamController。取消订阅会关闭两端。输入流和输出流可以独立使用（通常只需要服务器端流）。必要时，方法可以接受多个流参数。请明确 API 结构，并在修改后重新生成。

生命周期

• 每次调用都会创建一个服务器 Session；对于返回流的方法，会话会一直保持活动状态，直到返回的流完成或被取消。
• WebSocket 断开连接会关闭两端的流。
• 返回流的方法会一直保持活动状态，直到被取消或完成。

错误处理

抛出可序列化的异常会关闭流；客户端会在 onError 中收到该异常。异常可以双向传递。在 .spy.yaml 中定义异常类型。

与服务器事件结合的示例（简化版）

class PixelDrawingEndpoint extends Endpoint {
  static const _channelPixelAdded = 'pixel-added';
  final _pixelData = Uint8List(_numPixels);

  Future<void> setPixel(
    Session session, {
    required int colorIndex,
    required int pixelIndex,
  }) async {
    _pixelData[pixelIndex] = colorIndex;

    // Notify all connected clients that we set a pixel, by posting a message
    // to the _channelPixelAdded channel.
    session.messages.postMessage(
      _channelPixelAdded,
      ImageUpdate(
        pixelIndex: pixelIndex,
        colorIndex: colorIndex,
      ),
    );
  }

  /// Returns a stream of image updates. The first message will always be a
  /// `ImageData` object, which contains the full image. Sequential updates
  /// will be `ImageUpdate` objects, which contains a single updated pixel.
  Stream imageUpdates(Session session) async* {
    var updateStream =
        session.messages.createStream<ImageUpdate>(_channelPixelAdded);

    yield ImageData(
      pixels: _pixelData.buffer.asByteData(),
      width: _imageWidth,
      height: _imageHeight,
    );

    await for (var imageUpdate in updateStream) {
      yield imageUpdate;
    }
  }
}