11use super :: misc:: { marshal, unmarshal} ;
22use bytes:: BytesMut ;
33use rsocket_rust:: error:: RSocketError ;
4- use rsocket_rust:: extension:: { CompositeMetadata , MimeType , RoutingMetadata } ;
4+ use rsocket_rust:: extension:: {
5+ CompositeMetadata , CompositeMetadataEntry , MimeType , RoutingMetadata ,
6+ } ;
57use rsocket_rust:: prelude:: * ;
68use rsocket_rust:: utils:: Writeable ;
9+ use rsocket_rust_transport_tcp:: TcpClientTransport ;
10+ use rsocket_rust_transport_websocket:: WebsocketClientTransport ;
711use serde:: { Deserialize , Serialize } ;
812use std:: collections:: LinkedList ;
913use std:: error:: Error ;
14+ use std:: net:: SocketAddr ;
15+ use std:: result:: Result ;
1016use std:: sync:: Arc ;
17+ use url:: Url ;
18+
19+ enum TransportKind {
20+ TCP ( String , u16 ) ,
21+ WS ( String ) ,
22+ }
1123
1224pub struct Requester {
1325 rsocket : Arc < Box < dyn RSocket > > ,
@@ -20,6 +32,159 @@ pub struct RequestSpec {
2032 metadatas : LinkedList < ( MimeType , Vec < u8 > ) > ,
2133}
2234
35+ pub struct RequesterBuilder {
36+ data_mime_type : Option < MimeType > ,
37+ route : Option < String > ,
38+ metadata : Vec < CompositeMetadataEntry > ,
39+ data : Option < Vec < u8 > > ,
40+ tp : Option < TransportKind > ,
41+ }
42+
43+ impl Default for RequesterBuilder {
44+ fn default ( ) -> Self {
45+ Self {
46+ data_mime_type : None ,
47+ route : None ,
48+ metadata : Default :: default ( ) ,
49+ data : None ,
50+ tp : None ,
51+ }
52+ }
53+ }
54+
55+ impl RequesterBuilder {
56+ pub fn data_mime_type < I > ( mut self , mime_type : I ) -> Self
57+ where
58+ I : Into < MimeType > ,
59+ {
60+ self . data_mime_type = Some ( mime_type. into ( ) ) ;
61+ self
62+ }
63+
64+ pub fn setup_route < I > ( mut self , route : I ) -> Self
65+ where
66+ I : Into < String > ,
67+ {
68+ self . route = Some ( route. into ( ) ) ;
69+ self
70+ }
71+
72+ pub fn setup_data < D > ( mut self , data : & D ) -> Self
73+ where
74+ D : Sized + Serialize ,
75+ {
76+ // TODO: lazy set
77+ let mut bf = BytesMut :: new ( ) ;
78+ let result = match & self . data_mime_type {
79+ Some ( m) => marshal ( m, & mut bf, data) ,
80+ None => marshal ( & MimeType :: APPLICATION_JSON , & mut bf, data) ,
81+ } ;
82+ match result {
83+ Ok ( ( ) ) => {
84+ self . data = Some ( bf. to_vec ( ) ) ;
85+ }
86+ Err ( e) => {
87+ error ! ( "marshal failed: {:?}" , e) ;
88+ }
89+ }
90+ self
91+ }
92+
93+ pub fn setup_metadata < M , T > ( mut self , metadata : & M , mime_type : T ) -> Self
94+ where
95+ M : Sized + Serialize ,
96+ T : Into < MimeType > ,
97+ {
98+ // TODO: lazy set
99+ let mut bf = BytesMut :: new ( ) ;
100+ let mime_type = mime_type. into ( ) ;
101+ match marshal ( & mime_type, & mut bf, metadata) {
102+ Ok ( ( ) ) => {
103+ let entry = CompositeMetadataEntry :: new ( mime_type, bf. freeze ( ) ) ;
104+ self . metadata . push ( entry) ;
105+ }
106+ Err ( e) => error ! ( "marshal failed: {:?}" , e) ,
107+ }
108+ self
109+ }
110+
111+ pub fn connect_tcp < A > ( mut self , host : A , port : u16 ) -> Self
112+ where
113+ A : Into < String > ,
114+ {
115+ self . tp = Some ( TransportKind :: TCP ( host. into ( ) , port) ) ;
116+ self
117+ }
118+
119+ pub fn connect_websocket < I > ( mut self , url : I ) -> Self
120+ where
121+ I : Into < String > ,
122+ {
123+ self . tp = Some ( TransportKind :: WS ( url. into ( ) ) ) ;
124+ self
125+ }
126+
127+ pub async fn build ( self ) -> Result < Requester , Box < dyn Error + Send + Sync > > {
128+ let data_mime_type = self . data_mime_type . unwrap_or ( MimeType :: APPLICATION_JSON ) ;
129+
130+ let mut added = 0usize ;
131+ let mut composite_builder = CompositeMetadata :: builder ( ) ;
132+
133+ if let Some ( s) = self . route {
134+ let routing = RoutingMetadata :: builder ( ) . push ( s) . build ( ) ;
135+ composite_builder =
136+ composite_builder. push ( MimeType :: MESSAGE_X_RSOCKET_ROUTING_V0 , routing. bytes ( ) ) ;
137+ added += 1 ;
138+ }
139+ for it in self . metadata . into_iter ( ) {
140+ composite_builder = composite_builder. push_entry ( it) ;
141+ added += 1 ;
142+ }
143+
144+ let mut payload_builder = Payload :: builder ( ) ;
145+
146+ if added > 0 {
147+ payload_builder = payload_builder. set_metadata ( composite_builder. build ( ) ) ;
148+ }
149+
150+ if let Some ( raw) = self . data {
151+ payload_builder = payload_builder. set_data ( raw) ;
152+ }
153+
154+ let setup = payload_builder. build ( ) ;
155+
156+ match self . tp {
157+ Some ( TransportKind :: TCP ( h, p) ) => {
158+ let addr: SocketAddr = format ! ( "{}:{}" , h, p) . parse ( ) ?;
159+ let cli = RSocketFactory :: connect ( )
160+ . data_mime_type ( data_mime_type. as_ref ( ) )
161+ . setup ( setup)
162+ . metadata_mime_type ( MimeType :: MESSAGE_X_RSOCKET_COMPOSITE_METADATA_V0 . as_ref ( ) )
163+ . transport ( TcpClientTransport :: from ( addr) )
164+ . start ( )
165+ . await ?;
166+ let rsocket: Box < dyn RSocket > = Box :: new ( cli) ;
167+ let requester = Requester :: from ( rsocket) ;
168+ Ok ( requester)
169+ }
170+ Some ( TransportKind :: WS ( u) ) => {
171+ let url = Url :: parse ( & u) ?;
172+ let cli = RSocketFactory :: connect ( )
173+ . data_mime_type ( data_mime_type. as_ref ( ) )
174+ . setup ( setup)
175+ . metadata_mime_type ( MimeType :: MESSAGE_X_RSOCKET_COMPOSITE_METADATA_V0 . as_ref ( ) )
176+ . transport ( WebsocketClientTransport :: from ( url) )
177+ . start ( )
178+ . await ?;
179+ let rsocket: Box < dyn RSocket > = Box :: new ( cli) ;
180+ let requester = Requester :: from ( rsocket) ;
181+ Ok ( requester)
182+ }
183+ None => Err ( "Missing transport!" . into ( ) ) ,
184+ }
185+ }
186+ }
187+
23188impl From < Box < dyn RSocket > > for Requester {
24189 fn from ( rsocket : Box < dyn RSocket > ) -> Requester {
25190 Requester {
@@ -29,6 +194,10 @@ impl From<Box<dyn RSocket>> for Requester {
29194}
30195
31196impl Requester {
197+ pub fn builder ( ) -> RequesterBuilder {
198+ RequesterBuilder :: default ( )
199+ }
200+
32201 pub fn route ( & self , route : & str ) -> RequestSpec {
33202 let routing = RoutingMetadata :: builder ( ) . push_str ( route) . build ( ) ;
34203 let mut buf = BytesMut :: new ( ) ;
0 commit comments