fluvio-spu 0.2.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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

use tracing::trace;
use tracing::debug;
use futures_util::io::AsyncRead;
use futures_util::io::AsyncWrite;

use fluvio_socket::InnerFlvSink;
use fluvio_socket::InnerExclusiveFlvSink;
use fluvio_socket::FlvSocketError;
use dataplane::api::RequestMessage;
use dataplane::fetch::{FileFetchResponse, FileFetchRequest, FilePartitionResponse, FileTopicResponse};
use fluvio_controlplane_metadata::partition::ReplicaKey;
use fluvio_future::zero_copy::ZeroCopyWrite;

use crate::core::DefaultSharedGlobalContext;

/// perform log fetch request using zero copy write
pub async fn handle_fetch_request<S>(
    request: RequestMessage<FileFetchRequest>,
    ctx: DefaultSharedGlobalContext,
    sink: InnerExclusiveFlvSink<S>,
) -> Result<(), FlvSocketError>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
    InnerFlvSink<S>: ZeroCopyWrite,
{
    let (header, fetch_request) = request.get_header_request();
    let mut fetch_response = FileFetchResponse::default();

    for topic_request in &fetch_request.topics {
        let topic = &topic_request.name;

        let mut topic_response = FileTopicResponse::default();
        topic_response.name = topic.clone();

        for partition_req in &topic_request.fetch_partitions {
            let partition = &partition_req.partition_index;
            debug!(
                "fetch log: {}-{}, max_bytes: {}",
                topic, partition, fetch_request.max_bytes
            );
            let fetch_offset = partition_req.fetch_offset;
            let rep_id = ReplicaKey::new(topic.clone(), *partition);
            let mut partition_response = FilePartitionResponse::default();
            partition_response.partition_index = *partition;

            ctx.leaders_state()
                .read_records(
                    &rep_id,
                    fetch_offset,
                    fetch_request.max_bytes as u32,
                    fetch_request.isolation_level.clone(),
                    &mut partition_response,
                )
                .await;

            topic_response.partitions.push(partition_response);
        }

        fetch_response.topics.push(topic_response);
    }

    let response =
        RequestMessage::<FileFetchRequest>::response_with_header(&header, fetch_response);
    trace!("sending back file fetch response: {:#?}", response);
    let mut inner = sink.lock().await;
    inner
        .encode_file_slices(&response, header.api_version())
        .await?;
    drop(inner);
    trace!("finish sending fetch response");

    Ok(())
}