Grcov report - dldata

1

use std::error::Error as StdError;

2

3

use axum::{

4

    body::Body,

5

    extract::{Path, Request, State},

6

    response::IntoResponse,

7

    routing, Router,

8

};

9

use bytes::{Bytes, BytesMut};

10

use csv::WriterBuilder;

11

use futures_util::StreamExt;

12

use log::error;

13

use serde::{Deserialize, Serialize};

14

use serde_json::Deserializer;

15

16

use sylvia_iot_corelib::err::ErrResp;

17

18

use super::{super::State as AppState, api_bridge, list_api_bridge, ListResp};

19

20

#[derive(Deserialize)]

21

struct DataIdPath {

22

    data_id: String,

23

24

25

#[derive(Deserialize, Serialize)]

26

struct DlDataBuffer {

27

    #[serde(rename = "dataId")]

28

    data_id: String,

29

    #[serde(rename = "unitId")]

30

    unit_id: String,

31

    #[serde(rename = "applicationId")]

32

    application_id: String,

33

    #[serde(rename = "applicationCode")]

34

    application_code: String,

35

    #[serde(rename = "deviceId")]

36

    device_id: String,

37

    #[serde(rename = "networkId")]

38

    network_id: String,

39

    #[serde(rename = "createdAt")]

40

    created_at: String,

41

    #[serde(rename = "expiredAt")]

42

    expired_at: String,

43

44

45

const CSV_FIELDS: &'static [u8] =

46

    b"\xEF\xBB\xBFdataId,unitId,applicationId,applicationCode,deviceId,networkId,createdAt,expiredAt\n";

47

48

253

pub fn new_service(scope_path: &str, state: &AppState) -> Router {

49

253

    Router::new().nest(

50

253

        scope_path,

51

253

        Router::new()

52

253

            .route("/count", routing::get(get_dldata_buffer_count))

53

253

            .route("/list", routing::get(get_dldata_buffer_list))

54

253

            .route("/:data_id", routing::delete(delete_dldata_buffer))

55

253

            .with_state(state.clone()),

56

253

57

253

58

59

/// `GET /{base}/api/v1/dldata-buffer/count`

60

2

async fn get_dldata_buffer_count(state: State<AppState>, req: Request) -> impl IntoResponse {

61

    const FN_NAME: &'static str = "get_dldata_buffer_count";

62

2

    let api_path = format!("{}/api/v1/dldata-buffer/count", state.broker_base.as_str());

63

2

    let client = state.client.clone();

64

2

65

2

    api_bridge(FN_NAME, &client, req, api_path.as_str()).await

66

2

67

68

/// `GET /{base}/api/v1/dldata-buffer/list`

69

10

async fn get_dldata_buffer_list(state: State<AppState>, req: Request) -> impl IntoResponse {

70

    const FN_NAME: &'static str = "get_dldata_buffer_list";

71

10

    let api_path = format!("{}/api/v1/dldata-buffer/list", state.broker_base.as_str());

72

10

    let api_path = api_path.as_str();

73

10

    let client = state.client.clone();

74

75

2

    let (api_resp, resp_builder) =

76

10

        match list_api_bridge(FN_NAME, &client, req, api_path, false, "dldata-buffer").await {

77

8

            ListResp::Axum(resp) => return resp,

78

2

            ListResp::ArrayStream(api_resp, resp_builder) => (api_resp, resp_builder),

79

2

};

80

2

81

2

    let mut resp_stream = api_resp.bytes_stream();

82

2

    let body = Body::from_stream(async_stream::stream! {

83

2

        yield Ok(Bytes::from(CSV_FIELDS));

84

2

85

2

        let mut buffer = BytesMut::new();

86

2

        while let Some(body) = resp_stream.next().await {

87

2

            match body {

88

2

                Err(e) => {

89

2

                    error!("[{}] get body error: {}", FN_NAME, e);

90

2

                    let err: Box<dyn StdError + Send + Sync> = Box::new(e);

91

2

                    yield Err(err);

92

2

                    break;

93

2

94

2

                Ok(body) => buffer.extend_from_slice(&body[..]),

95

2

96

2

97

2

            let mut json_stream =

98

2

                Deserializer::from_slice(&buffer[..]).into_iter::<DlDataBuffer>();

99

2

            let mut index = 0;

100

2

            let mut finish = false;

101

2

            loop {

102

2

                if let Some(Ok(v)) = json_stream.next() {

103

2

                    let mut writer = WriterBuilder::new().has_headers(false).from_writer(vec![]);

104

2

                    if let Err(e) = writer.serialize(v) {

105

2

                        let err: Box<dyn StdError + Send + Sync> = Box::new(e);

106

2

                        yield Err(err);

107

2

                        finish = true;

108

2

                        break;

109

2

110

2

                    match writer.into_inner() {

111

2

                        Err(e) => {

112

2

                            let err: Box<dyn StdError + Send + Sync> = Box::new(e);

113

2

                            yield Err(err);

114

2

                            finish = true;

115

2

                            break;

116

2

117

2

                        Ok(row) => yield Ok(Bytes::copy_from_slice(row.as_slice())),

118

2

119

2

                    continue;

120

2

121

2

                let offset = json_stream.byte_offset();

122

2

                if buffer.len() <= index + offset {

123

2

                    index = buffer.len();

124

2

                    break;

125

2

126

2

                match buffer[index+offset] {

127

2

                    b'[' | b',' => {

128

2

                        index += offset + 1;

129

2

                        if buffer.len() <= index {

130

2

                            break;

131

2

132

2

                        json_stream =

133

2

                            Deserializer::from_slice(&buffer[index..]).into_iter::<DlDataBuffer>();

134

2

135

2

                    b']' => {

136

2

                        finish = true;

137

2

                        break;

138

2

139

2

                    _ => break,

140

2

141

2

142

2

            if finish {

143

2

                break;

144

2

145

2

            buffer = buffer.split_off(index);

146

2

147

2

});

148

2

    match resp_builder.body(body) {

149

        Err(e) => ErrResp::ErrRsc(Some(e.to_string())).into_response(),

150

2

        Ok(resp) => resp,

151

152

10

153

154

/// `DELETE /{base}/api/v1/dldata-buffer/{dataId}`

155

2

async fn delete_dldata_buffer(

156

2

    state: State<AppState>,

157

2

    Path(param): Path<DataIdPath>,

158

2

    req: Request,

159

2

) -> impl IntoResponse {

160

    const FN_NAME: &'static str = "delete_dldata_buffer";

161

2

    let api_path = format!(

162

2

        "{}/api/v1/dldata-buffer/{}",

163

2

        state.broker_base, param.data_id

164

2

);

165

2

    let client = state.client.clone();

166

2

167

2

    api_bridge(FN_NAME, &client, req, api_path.as_str()).await

168

2