5829059745

pending completion

Build # 5829059745

Build Type

Pull #1844

github

Committed by

supergi01

Commit Message

added comments for downloader.rs

Pull Request Pull Request #1844: feat/live-reload, download and start react server

Run Details

735 of 735 new or added lines in 11 files covered. (100.0%)

45536 of 61287 relevant lines covered (74.3%)

51206.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/dozer-cli/src/live/server.rs

use std::sync::Arc;

use dozer_api::{tonic_reflection, tonic_web};
use dozer_types::{
    grpc_types::{
        live::{
            code_service_server::{CodeService, CodeServiceServer},
            CommonRequest, CommonResponse, DotResponse, LiveResponse, RunSqlRequest,
            SchemasResponse, SourcesRequest, SqlRequest, SqlResponse,
        },
        types::Operation,
    },
    log::{error, info},
};
use futures::stream::BoxStream;
use tokio::sync::broadcast::Receiver;

use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status};

use super::state::LiveState;
const LIVE_PORT: u16 = 4556;
pub struct LiveServer {
    pub receiver: Receiver<LiveResponse>,
    pub state: Arc<LiveState>,
}

#[tonic::async_trait]
impl CodeService for LiveServer {
    type LiveConnectStream = BoxStream<'static, Result<LiveResponse, Status>>;
    type RunSqlStream = BoxStream<'static, Result<Operation, Status>>;

    async fn live_connect(
        &self,
        _request: Request<CommonRequest>,
    ) -> Result<Response<Self::LiveConnectStream>, Status> {
        let (tx, rx) = tokio::sync::mpsc::channel(1);

        let mut receiver = self.receiver.resubscribe();
        let initial_state = self.state.clone();
        tokio::spawn(async move {
            let initial_state = initial_state.get_current();
            match initial_state {
                Ok(initial_state) => {
                    if let Err(e) = tx.send(Ok(initial_state)).await {
                        info!("Error getting initial state");
                        info!("{:?}", e);
                    }
                }
                Err(e) => {
                    info!("Error sending to channel");
                    info!("{:?}", e);
                }
            };

            loop {
                let res = receiver.recv().await;
                match res {
                    Ok(res) => {
                        let res = tx.send(Ok(res)).await;
                        match res {
                            Ok(_) => {}
                            Err(e) => {
                                error!("Error sending to channel");
                                error!("{:?}", e);
                                break;
                            }
                        }
                    }
                    Err(_) => {
                        break;
                    }
                }
            }
        });
        let stream = ReceiverStream::new(rx);

        Ok(Response::new(Box::pin(stream) as Self::LiveConnectStream))
    }

    async fn sources(
        &self,
        request: Request<SourcesRequest>,
    ) -> Result<Response<SchemasResponse>, Status> {
        let req = request.into_inner();
        let res = self.state.get_source_schemas(req.connection_name).await;
        match res {
            Ok(res) => Ok(Response::new(res)),
            Err(e) => Err(Status::internal(e.to_string())),
        }
    }

    async fn endpoints(
        &self,
        _request: Request<CommonRequest>,
    ) -> Result<Response<SchemasResponse>, Status> {
        let state = self.state.clone();
        let handle = std::thread::spawn(move || state.get_endpoints_schemas());
        let res = handle.join().unwrap();

        match res {
            Ok(res) => Ok(Response::new(res)),
            Err(e) => Err(Status::internal(e.to_string())),
        }
    }

    async fn generate_dot(
        &self,
        _request: Request<CommonRequest>,
    ) -> Result<Response<DotResponse>, Status> {
        let state = self.state.clone();
        let handle = std::thread::spawn(move || state.generate_dot());
        let res = handle.join().unwrap();

        match res {
            Ok(res) => Ok(Response::new(res)),
            Err(e) => Err(Status::internal(e.to_string())),
        }
    }

    async fn get_sql(
        &self,
        _request: Request<CommonRequest>,
    ) -> Result<Response<SqlResponse>, Status> {
        let res = self.state.get_sql();

        match res {
            Ok(res) => Ok(Response::new(res)),
            Err(e) => Err(Status::internal(e.to_string())),
        }
    }

    async fn build_sql(
        &self,
        request: Request<SqlRequest>,
    ) -> Result<Response<SchemasResponse>, Status> {
        let state = self.state.clone();
        let handle = std::thread::spawn(move || state.build_sql(request.into_inner().sql));
        let res = handle.join().unwrap();

        match res {
            Ok(res) => Ok(Response::new(res)),
            Err(e) => Err(Status::internal(e.to_string())),
        }
    }

    async fn run_sql(
        &self,
        request: Request<RunSqlRequest>,
    ) -> Result<Response<Self::RunSqlStream>, Status> {
        let (tx, rx) = tokio::sync::mpsc::channel(1);

        let req = request.into_inner();
        let state = self.state.clone();

        state.run_sql(req.sql, req.endpoints, tx).unwrap();

        let stream = ReceiverStream::new(rx);

        Ok(Response::new(Box::pin(stream) as Self::RunSqlStream))
    }

    async fn stop_sql(
        &self,
        _request: Request<CommonRequest>,
    ) -> Result<Response<CommonResponse>, Status> {
        self.state.stop_sql();
        Ok(Response::new(CommonResponse {}))
    }
}

pub async fn serve(
    receiver: Receiver<LiveResponse>,
    state: Arc<LiveState>,
) -> Result<(), tonic::transport::Error> {
    let addr = format!("0.0.0.0:{LIVE_PORT}").parse().unwrap();
    let live_server = LiveServer { receiver, state };
    let svc = CodeServiceServer::new(live_server);

    // Enable CORS for local development
    let svc = tonic_web::config().allow_all_origins().enable(svc);

    let reflection_service = tonic_reflection::server::Builder::configure()
        .register_encoded_file_descriptor_set(dozer_types::grpc_types::live::FILE_DESCRIPTOR_SET)
        .build()
        .unwrap();

    tonic::transport::Server::builder()
        .accept_http1(true)
        .add_service(svc)
        .add_service(reflection_service)
        .serve(addr)
        .await
}

1	use std::sync::Arc;
2
3	use dozer_api::{tonic_reflection, tonic_web};
4	use dozer_types::{
5	grpc_types::{
6	live::{
7	code_service_server::{CodeService, CodeServiceServer},
8	CommonRequest, CommonResponse, DotResponse, LiveResponse, RunSqlRequest,
9	SchemasResponse, SourcesRequest, SqlRequest, SqlResponse,
10	},
11	types::Operation,
12	},
13	log::{error, info},
14	};
15	use futures::stream::BoxStream;
16	use tokio::sync::broadcast::Receiver;
17
18	use tokio_stream::wrappers::ReceiverStream;
19	use tonic::{Request, Response, Status};
20
21	use super::state::LiveState;
22	const LIVE_PORT: u16 = 4556;
23	pub struct LiveServer {
24	pub receiver: Receiver<LiveResponse>,
25	pub state: Arc<LiveState>,
26	}
27
28	#[tonic::async_trait]
29	impl CodeService for LiveServer {
30	type LiveConnectStream = BoxStream<'static, Result<LiveResponse, Status>>;
31	type RunSqlStream = BoxStream<'static, Result<Operation, Status>>;
32		×
33	async fn live_connect(	×
34	&self,	×
35	_request: Request<CommonRequest>,	×
36	) -> Result<Response<Self::LiveConnectStream>, Status> {	×
37	let (tx, rx) = tokio::sync::mpsc::channel(1);	×
38		×
39	let mut receiver = self.receiver.resubscribe();	×
40	let initial_state = self.state.clone();	×
41	tokio::spawn(async move {	×
42	let initial_state = initial_state.get_current();	×
43	match initial_state {	×
44	Ok(initial_state) => {	×
45	if let Err(e) = tx.send(Ok(initial_state)).await {	×
46	info!("Error getting initial state");	×
47	info!("{:?}", e);	×
48	}	×
49	}	×
50	Err(e) => {	×
51	info!("Error sending to channel");	×
52	info!("{:?}", e);	×
53	}
54	};
55
56	loop {	×
57	let res = receiver.recv().await;	×
58	match res {	×
59	Ok(res) => {	×
60	let res = tx.send(Ok(res)).await;	×
61	match res {	×
62	Ok(_) => {}	×
63	Err(e) => {	×
64	error!("Error sending to channel");	×
65	error!("{:?}", e);	×
66	break;	×
67	}
68	}
69	}
70	Err(_) => {	×
71	break;	×
72	}
73	}
74	}	×
75	});	×
76	let stream = ReceiverStream::new(rx);	×
77		×
78	Ok(Response::new(Box::pin(stream) as Self::LiveConnectStream))	×
79	}	×
80		×
81	async fn sources(	×
82	&self,	×
83	request: Request<SourcesRequest>,	×
84	) -> Result<Response<SchemasResponse>, Status> {	×
85	let req = request.into_inner();	×
86	let res = self.state.get_source_schemas(req.connection_name).await;	×
87	match res {	×
88	Ok(res) => Ok(Response::new(res)),	×
89	Err(e) => Err(Status::internal(e.to_string())),	×
90	}	×
91	}	×
92		×
93	async fn endpoints(	×
94	&self,	×
95	_request: Request<CommonRequest>,	×
96	) -> Result<Response<SchemasResponse>, Status> {	×
97	let state = self.state.clone();	×
98	let handle = std::thread::spawn(move \|\| state.get_endpoints_schemas());	×
99	let res = handle.join().unwrap();	×
100		×
101	match res {	×
102	Ok(res) => Ok(Response::new(res)),	×
103	Err(e) => Err(Status::internal(e.to_string())),	×
104	}	×
105	}	×
106		×
107	async fn generate_dot(	×
108	&self,	×
109	_request: Request<CommonRequest>,	×
110	) -> Result<Response<DotResponse>, Status> {	×
111	let state = self.state.clone();	×
112	let handle = std::thread::spawn(move \|\| state.generate_dot());	×
113	let res = handle.join().unwrap();	×
114		×
115	match res {	×
116	Ok(res) => Ok(Response::new(res)),	×
117	Err(e) => Err(Status::internal(e.to_string())),	×
118	}	×
119	}	×
120		×
121	async fn get_sql(	×
122	&self,	×
123	_request: Request<CommonRequest>,	×
124	) -> Result<Response<SqlResponse>, Status> {	×
125	let res = self.state.get_sql();	×
126		×
127	match res {	×
128	Ok(res) => Ok(Response::new(res)),	×
129	Err(e) => Err(Status::internal(e.to_string())),	×
130	}	×
131	}	×
132		×
133	async fn build_sql(	×
134	&self,	×
135	request: Request<SqlRequest>,	×
136	) -> Result<Response<SchemasResponse>, Status> {	×
137	let state = self.state.clone();	×
138	let handle = std::thread::spawn(move \|\| state.build_sql(request.into_inner().sql));	×
139	let res = handle.join().unwrap();	×
140		×
141	match res {	×
142	Ok(res) => Ok(Response::new(res)),	×
143	Err(e) => Err(Status::internal(e.to_string())),	×
144	}	×
145	}	×
146		×
147	async fn run_sql(	×
148	&self,	×
149	request: Request<RunSqlRequest>,	×
150	) -> Result<Response<Self::RunSqlStream>, Status> {	×
151	let (tx, rx) = tokio::sync::mpsc::channel(1);	×
152		×
153	let req = request.into_inner();	×
154	let state = self.state.clone();	×
155		×
156	state.run_sql(req.sql, req.endpoints, tx).unwrap();	×
157		×
158	let stream = ReceiverStream::new(rx);	×
159		×
160	Ok(Response::new(Box::pin(stream) as Self::RunSqlStream))	×
161	}	×
162
163	async fn stop_sql(	×
164	&self,	×
165	_request: Request<CommonRequest>,	×
166	) -> Result<Response<CommonResponse>, Status> {	×
167	self.state.stop_sql();	×
168	Ok(Response::new(CommonResponse {}))	×
169	}	×
170	}	×
171		×
172	pub async fn serve(	×
173	receiver: Receiver<LiveResponse>,	×
174	state: Arc<LiveState>,	×
175	) -> Result<(), tonic::transport::Error> {	×
176	let addr = format!("0.0.0.0:{LIVE_PORT}").parse().unwrap();	×
177	let live_server = LiveServer { receiver, state };	×
178	let svc = CodeServiceServer::new(live_server);	×
179		×
180	// Enable CORS for local development	×
181	let svc = tonic_web::config().allow_all_origins().enable(svc);	×
182		×
183	let reflection_service = tonic_reflection::server::Builder::configure()	×
184	.register_encoded_file_descriptor_set(dozer_types::grpc_types::live::FILE_DESCRIPTOR_SET)	×
185	.build()	×
186	.unwrap();	×
187		×
188	tonic::transport::Server::builder()	×
189	.accept_http1(true)	×
190	.add_service(svc)	×
191	.add_service(reflection_service)	×
192	.serve(addr)	×
193	.await	×
194	}	×

getdozer / dozer / 5829059745

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous