serverpod-streams

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Serverpod Streams

Serverpod 流处理

Endpoint methods that take or return

Stream<T>

get WebSocket-managed client stubs. Types must be serializable.

It's a common use case to use streams together with server events. See Server Events.

接收或返回

Stream<T>

的Endpoint方法会生成由WebSocket管理的客户端桩代码。类型必须可序列化。

流与服务器事件结合是常见的使用场景。请查看Server Events。

Defining a streaming method

定义流方法

dart

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

Run

serverpod generate

for client stubs.

dart

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

运行

serverpod generate

生成客户端桩代码。

Client usage

客户端使用

dart

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

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

Close the

StreamController

when done. Cancelling the subscription closes both sides. We can use the in/out stream independently (often only need for stream from server). Methods may take multiple stream parameters when needed. Keep the API shape explicit and regenerate after changes.

dart

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

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

使用完毕后关闭

StreamController

。取消订阅会关闭两端的流。我们可以独立使用输入/输出流（通常只需要来自服务器的流）。必要时方法可以接收多个流参数。保持API结构清晰，修改后重新生成代码。

Lifecycle

生命周期

Each call creates a server
```
Session
```
; for return-stream methods, the session stays alive until the returned stream completes or is cancelled.
WebSocket disconnect closes streams on both sides.
Return-stream methods stay alive until cancelled or completed.

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

Error handling

错误处理

Throwing a serializable exception closes the stream; the client receives it in

onError

. Exceptions can flow in both directions. Define exception types in

.spy.yaml

抛出可序列化的异常会关闭流；客户端会在

onError

中接收该异常。异常可以在双向流动。在

.spy.yaml

中定义异常类型。

Example combined with server events (simplified)

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

dart

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;
    }
  }
}

dart

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;
    }
  }
}