流式传输

syntax = "proto3";
package streaming.v1;
service StreamingService {
  // 服务器流式传输：向客户端发送多个事件
  rpc StreamUserEvents(StreamUserEventsRequest) returns (stream UserEvent);
  
  // 服务器流式传输：分块下载大文件
  rpc DownloadFile(DownloadFileRequest) returns (stream FileChunk);
  
  // 服务器流式传输：实时通知
  rpc SubscribeToNotifications(SubscribeRequest) returns (stream Notification);
}
message StreamUserEventsRequest {
  string user_id = 1;
  repeated UserEventType event_types = 2;
  google.protobuf.Timestamp start_time = 3;
}
message UserEvent {
  string id = 1;
  string user_id = 2;
  UserEventType type = 3;
  google.protobuf.Timestamp timestamp = 4;
  google.protobuf.Any data = 5;
}
enum UserEventType {
  USER_EVENT_TYPE_UNSPECIFIED = 0;
  USER_EVENT_TYPE_LOGIN = 1;
  USER_EVENT_TYPE_LOGOUT = 2;
  USER_EVENT_TYPE_PROFILE_UPDATE = 3;
  USER_EVENT_TYPE_PASSWORD_CHANGE = 4;
}

import grpc
import time
import asyncio
from grpc import ServicerContext
from typing import Iterator
class StreamingServiceServicer(streaming_pb2_grpc.StreamingServiceServicer):
    
    def StreamUserEvents(
        self, 
        request: streaming_pb2.StreamUserEventsRequest, 
        context: ServicerContext
    ) -> Iterator[streaming_pb2.UserEvent]:
        """实时流式传输用户事件。"""
        
        # 验证请求
        if not request.user_id:
            context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
            context.set_details('User ID is required')
            return
        
        # 订阅事件流
        event_subscriber = self.event_store.subscribe(
            user_id=request.user_id,
            event_types=request.event_types,
            start_time=request.start_time
        )
        
        try:
            while context.is_active():
                # 等待下一个事件（带超时）
                try:
                    event = event_subscriber.get_next_event(timeout=30)
                    if event:
                        yield event
                    else:
                        # 发送心跳以保持连接活跃
                        continue
                        
                except TimeoutError:
                    # 检查客户端是否仍然连接
                    if not context.is_active():
                        break
                        
                except Exception as e:
                    context.set_code(grpc.StatusCode.INTERNAL)
                    context.set_details(f'Error streaming events: {str(e)}')
                    break
                    
        finally:
            # 清理订阅
            event_subscriber.close()
    
    def DownloadFile(
        self, 
        request: streaming_pb2.DownloadFileRequest, 
        context: ServicerContext
    ) -> Iterator[streaming_pb2.FileChunk]:
        """分块下载文件。"""
        
        try:
            file_path = self.file_store.get_file_path(request.file_id)
            if not file_path or not os.path.exists(file_path):
                context.set_code(grpc.StatusCode.NOT_FOUND)
                context.set_details(f'File {request.file_id} not found')
                return
            
            chunk_size = 64 * 1024  # 64KB 块大小
            
            with open(file_path, 'rb') as f:
                while True:
                    chunk_data = f.read(chunk_size)
                    if not chunk_data:
                        break
                    
                    # 检查客户端是否断开连接
                    if not context.is_active():
                        break
                    
                    yield streaming_pb2.FileChunk(
                        data=chunk_data,
                        offset=f.tell() - len(chunk_data),
                        size=len(chunk_data)
                    )
                    
        except Exception as e:
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(f'Error downloading file: {str(e)}')

service StreamingService {
  // 客户端流式传输：分块上传大文件
  rpc UploadFile(stream FileChunk) returns (UploadFileResponse);
  
  // 客户端流式传输：批量数据处理
  rpc ProcessDataBatch(stream DataRecord) returns (BatchProcessingResult);
  
  // 客户端流式传输：实时指标收集
  rpc CollectMetrics(stream MetricData) returns (MetricsCollectionResult);
}
message FileChunk {
  oneof data {
    FileMetadata metadata = 1;
    bytes chunk = 2;
  }
}
message FileMetadata {
  string filename = 1;
  int64 total_size = 2;
  string content_type = 3;
  string checksum = 4;
}
message UploadFileResponse {
  string file_id = 1;
  int64 bytes_uploaded = 2;
  string download_url = 3;
  bool checksum_verified = 4;
}

class StreamingServiceServicer(streaming_pb2_grpc.StreamingServiceServicer):
    
    def UploadFile(
        self, 
        request_iterator: Iterator[streaming_pb2.FileChunk], 
        context: ServicerContext
    ) -> streaming_pb2.UploadFileResponse:
        """从客户端流上传文件。"""
        
        file_metadata = None
        total_bytes = 0
        file_path = None
        hasher = hashlib.sha256()
        
        try:
            for chunk in request_iterator:
                if chunk.HasField('metadata'):
                    # 第一个块包含元数据
                    file_metadata = chunk.metadata
                    
                    # 创建临时文件
                    file_id = str(uuid.uuid4())
                    file_path = f'/tmp/uploads/{file_id}'
                    os.makedirs(os.path.dirname(file_path), exist_ok=True)
                    
                elif chunk.HasField('chunk'):
                    # 后续块包含文件数据
                    if not file_metadata:
                        context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
                        context.set_details('File metadata must be sent first')
                        return streaming_pb2.UploadFileResponse()
                    
                    # 将块写入文件
                    with open(file_path, 'ab') as f:
                        f.write(chunk.chunk)
                    
                    total_bytes += len(chunk.chunk)
                    hasher.update(chunk.chunk)
                    
                    # 检查大小限制
                    if total_bytes > file_metadata.total_size:
                        context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
                        context.set_details('File size exceeds declared size')
                        return streaming_pb2.UploadFileResponse()
            
            # 验证校验和
            computed_checksum = hasher.hexdigest()
            checksum_verified = computed_checksum == file_metadata.checksum
            
            if not checksum_verified:
                context.set_code(grpc.StatusCode.DATA_LOSS)
                context.set_details('File checksum verification failed')
                return streaming_pb2.UploadFileResponse()
            
            # 将文件移动到永久存储
            permanent_path = self.file_store.store_file(file_id, file_path)
            download_url = self.file_store.get_download_url(file_id)
            
            return streaming_pb2.UploadFileResponse(
                file_id=file_id,
                bytes_uploaded=total_bytes,
                download_url=download_url,
                checksum_verified=True
            )
            
        except Exception as e:
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(f'Error uploading file: {str(e)}')
            return streaming_pb2.UploadFileResponse()
        
        finally:
            # 清理临时文件
            if file_path and os.path.exists(file_path):
                os.remove(file_path)

service StreamingService {
  // 双向流式传输：实时聊天
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
  
  // 双向流式传输：实时协作
  rpc Collaborate(stream CollaborationEvent) returns (stream CollaborationEvent);
  
  // 双向流式传输：实时数据处理
  rpc ProcessLiveData(stream DataInput) returns (stream ProcessingResult);
}
message ChatMessage {
  string id = 1;
  string user_id = 2;
  string room_id = 3;
  string content = 4;
  google.protobuf.Timestamp timestamp = 5;
  ChatMessageType type = 6;
}
enum ChatMessageType {
  CHAT_MESSAGE_TYPE_UNSPECIFIED = 0;
  CHAT_MESSAGE_TYPE_TEXT = 1;
  CHAT_MESSAGE_TYPE_IMAGE = 2;
  CHAT_MESSAGE_TYPE_FILE = 3;
  CHAT_MESSAGE_TYPE_SYSTEM = 4;
  CHAT_MESSAGE_TYPE_TYPING = 5;
}

import asyncio
import queue
import threading
class StreamingServiceServicer(streaming_pb2_grpc.StreamingServiceServicer):
    
    def Chat(
        self, 
        request_iterator: Iterator[streaming_pb2.ChatMessage], 
        context: ServicerContext
    ) -> Iterator[streaming_pb2.ChatMessage]:
        """双向聊天流式传输。"""
        
        # 出站消息队列
        outgoing_queue = queue.Queue()
        
        # 跟踪用户会话
        user_session = None
        
        def handle_incoming_messages():
            """处理来自客户端的入站消息。"""
            nonlocal user_session
            
            try:
                for message in request_iterator:
                    if not user_session:
                        # 第一条消息建立会话
                        user_session = self.chat_service.join_room(
                            user_id=message.user_id,
                            room_id=message.room_id
                        )
                        
                        # 发送欢迎消息
                        welcome_msg = streaming_pb2.ChatMessage(
                            id=str(uuid.uuid4()),
                            user_id="system",
                            room_id=message.room_id,
                            content=f"User {message.user_id} joined the chat",
                            timestamp=google.protobuf.timestamp_pb2.Timestamp(),
                            type=streaming_pb2.CHAT_MESSAGE_TYPE_SYSTEM
                        )
                        outgoing_queue.put(welcome_msg)
                    
                    # 处理消息
                    if message.type == streaming_pb2.CHAT_MESSAGE_TYPE_TEXT:
                        # 广播给房间内其他用户
                        self.chat_service.broadcast_message(message)
                        
                    elif message.type == streaming_pb2.CHAT_MESSAGE_TYPE_TYPING:
                        # 向其他用户发送打字指示器
                        self.chat_service.broadcast_typing(message)
                        
            except Exception as e:
                print(f"Error handling incoming messages: {e}")
            finally:
                # 清理会话
                if user_session:
                    self.chat_service.leave_room(user_session)
        
        # 启动处理入站消息的后台线程
        incoming_thread = threading.Thread(target=handle_incoming_messages)
        incoming_thread.daemon = True
        incoming_thread.start()
        
        # 订阅房间消息
        message_subscriber = None
        if user_session:
            message_subscriber = self.chat_service.subscribe_to_room(
                user_session.room_id,
                exclude_user=user_session.user_id
            )
        
        try:
            while context.is_active():
                # 检查队列中的出站消息
                try:
                    message = outgoing_queue.get(timeout=1)
                    yield message
                except queue.Empty:
                    pass
                
                # 检查来自其他用户的消息
                if message_subscriber:
                    try:
                        room_message = message_subscriber.get_message(timeout=1)
                        if room_message:
                            yield room_message
                    except TimeoutError:
                        pass
                        
        finally:
            # 清理
            if message_subscriber:
                message_subscriber.close()
            if user_session:
                self.chat_service.leave_room(user_session)

def StreamData(self, request, context):
    """带流量控制的流式传输。"""
    
    # 使用有界队列控制内存使用
    data_queue = queue.Queue(maxsize=100)
    
    def data_producer():
        """生成数据的后台线程。"""
        for item in self.data_source.get_items():
            try:
                data_queue.put(item, timeout=5)
            except queue.Full:
                # 应用反压
                print("Client is too slow, dropping data")
                break
    
    producer_thread = threading.Thread(target=data_producer)
    producer_thread.start()
    
    try:
        while context.is_active():
            try:
                item = data_queue.get(timeout=30)
                yield item
            except queue.Empty:
                # 发送心跳或检查客户端连接
                if not context.is_active():
                    break
    finally:
        producer_thread.join(timeout=1)

def StreamWithErrorHandling(self, request, context):
    """带健壮错误处理的流式传输。"""
    
    try:
        for item in self.get_stream_data(request):
            if not context.is_active():
                break
                
            try:
                # 处理项目
                processed_item = self.process_item(item)
                yield processed_item
                
            except ProcessingError as e:
                # 将错误作为响应的一部分发送
                error_response = create_error_response(e)
                yield error_response
                
            except Exception as e:
                # 严重错误 - 中止流
                context.set_code(grpc.StatusCode.INTERNAL)
                context.set_details(f'Processing failed: {str(e)}')
                break
                
    except Exception as e:
        context.set_code(grpc.StatusCode.INTERNAL)
        context.set_details(f'Stream failed: {str(e)}')

import grpc
def stream_chat_client():
    """双向流式传输客户端示例。"""
    
    channel = grpc.insecure_channel('localhost:50051')
    stub = streaming_pb2_grpc.StreamingServiceStub(channel)
    
    def message_generator():
        """生成出站消息。"""
        # 发送初始消息
        yield streaming_pb2.ChatMessage(
            user_id="user123",
            room_id="general",
            content="Hello, world!",
            type=streaming_pb2.CHAT_MESSAGE_TYPE_TEXT
        )
        
        # 保持连接活跃并定期发送消息
        while True:
            user_input = input("Enter message: ")
            if user_input.lower() == 'quit':
                break
                
            yield streaming_pb2.ChatMessage(
                user_id="user123",
                room_id="general", 
                content=user_input,
                type=streaming_pb2.CHAT_MESSAGE_TYPE_TEXT
            )
    
    # 启动双向流
    responses = stub.Chat(message_generator())
    
    try:
        for response in responses:
            print(f"Received: {response.content}")
    except grpc.RpcError as e:
        print(f"RPC failed: {e}")