fluvio-spu 0.3.0

Fluvio Stream Processing Unit
Documentation 
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
33
34

use tracing::debug;

use dataplane::api::RequestMessage;
use fluvio_socket::FlvSocket;
use fluvio_socket::FlvSocketError;

use crate::core::DefaultSharedGlobalContext;
use crate::controllers::leader_replica::LeaderConnection;
use super::FetchStreamRequest;
use super::FetchStreamResponse;

pub(crate) async fn handle_fetch_stream_request(
    req_msg: RequestMessage<FetchStreamRequest>,
    ctx: DefaultSharedGlobalContext,
    mut socket: FlvSocket,
) -> Result<(), FlvSocketError> {
    let request = &req_msg.request;
    let follower_id = request.spu_id;
    debug!(
        "internal service: respond to fetch stream request, follower: {}",
        follower_id
    );

    let response = FetchStreamResponse::new(follower_id);
    let res_msg = req_msg.new_response(response);
    socket
        .get_mut_sink()
        .send_response(&res_msg, req_msg.header.api_version())
        .await?;

    LeaderConnection::handle(ctx, follower_id, socket).await?;

    Ok(()) as Result<(), FlvSocketError>
}