Streaming APIs

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// WebSocket을 사용한 스트리밍 API 연결 예시
const socket = new WebSocket('wss://api.example.com/stream');

// 연결 수립 시 이벤트
socket.onopen = function(event) {
    console.log('스트리밍 연결이 수립되었습니다.');
    // 구독 요청 보내기
    socket.send(JSON.stringify({action: 'subscribe', channels: ['market-data']}));
};

// 데이터 수신 이벤트
socket.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log('실시간 데이터 수신:', data);
    // 수신된 데이터 처리 로직
};

// 오류 발생 시 이벤트
socket.onerror = function(error) {
    console.error('스트리밍 오류:', error);
};

// 연결 종료 시 이벤트
socket.onclose = function(event) {
    console.log('스트리밍 연결이 종료되었습니다:', event.code);
    // 재연결 로직
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// SSE를 사용한 스트리밍 API 연결 예시
const eventSource = new EventSource('https://api.example.com/events');

// 이벤트 수신
eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log('이벤트 데이터:', data);
    // 데이터 처리 로직
};

// 특정 이벤트 타입 수신
eventSource.addEventListener('market-update', function(event) {
    const marketData = JSON.parse(event.data);
    console.log('시장 데이터 업데이트:', marketData);
    // 시장 데이터 처리 로직
});

// 오류 처리
eventSource.onerror = function(error) {
    console.error('SSE 오류:', error);
    // 오류 처리 및 재연결 로직
};

// 연결 종료
function closeConnection() {
    eventSource.close();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// Fetch API를 사용한 HTTP 스트리밍 예시
fetch('https://api.example.com/stream', {
    method: 'GET',
    headers: {
        'Accept': 'application/json'
    }
})
.then(response => {
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    
    function processStream({ done, value }) {
        if (done) {
            console.log('스트림 완료');
            return;
        }
        
        // Uint8Array를 텍스트로 변환
        const chunk = decoder.decode(value, { stream: true });
        console.log('수신된 데이터 청크:', chunk);
        
        // 다음 청크 처리
        return reader.read().then(processStream);
    }
    
    return reader.read().then(processStream);
})
.catch(error => {
    console.error('스트리밍 오류:', error);
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# Python에서 gRPC 스트리밍 서비스 구현 예시
import grpc
from concurrent import futures
import time
import market_data_pb2
import market_data_pb2_grpc

class MarketDataService(market_data_pb2_grpc.MarketDataServiceServicer):
    def SubscribeToMarketData(self, request, context):
        # 요청된 심볼에 대한 시장 데이터 스트림 생성
        for symbol in request.symbols:
            # 실시간 데이터를 가져오는 로직
            while True:
                # 데이터 생성 또는 가져오기
                price = get_current_price(symbol)
                yield market_data_pb2.MarketDataResponse(
                    symbol=symbol,
                    price=price,
                    timestamp=int(time.time())
                )
                time.sleep(1)  # 1초마다 업데이트

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    market_data_pb2_grpc.add_MarketDataServiceServicer_to_server(
        MarketDataService(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()