@@ -35,8 +35,101 @@ pub enum LinkError {
3535 Elapsed ( #[ from] tokio:: time:: error:: Elapsed ) ,
3636}
3737
38+ // used to build LinkTx and LinkRx
39+ pub struct LinkBuilder < ' a > {
40+ tenant_id : Option < String > ,
41+ client_id : & ' a str ,
42+ router_tx : Sender < ( ConnectionId , Event ) > ,
43+ // true by default
44+ clean_session : bool ,
45+ last_will : Option < LastWill > ,
46+ // false by default
47+ dynamic_filters : bool ,
48+ // default to 0, indicating to not use topic alias
49+ topic_alias_max : u16 ,
50+ }
51+
52+ impl < ' a > LinkBuilder < ' a > {
53+ pub fn new ( client_id : & ' a str , router_tx : Sender < ( ConnectionId , Event ) > ) -> Self {
54+ LinkBuilder {
55+ client_id,
56+ router_tx,
57+ tenant_id : None ,
58+ clean_session : true ,
59+ last_will : None ,
60+ dynamic_filters : false ,
61+ topic_alias_max : 0 ,
62+ }
63+ }
64+
65+ pub fn tenant_id ( mut self , tenant_id : Option < String > ) -> Self {
66+ self . tenant_id = tenant_id;
67+ self
68+ }
69+
70+ pub fn last_will ( mut self , last_will : Option < LastWill > ) -> Self {
71+ self . last_will = last_will;
72+ self
73+ }
74+
75+ pub fn topic_alias_max ( mut self , max : u16 ) -> Self {
76+ self . topic_alias_max = max;
77+ self
78+ }
79+
80+ pub fn clean_session ( mut self , clean : bool ) -> Self {
81+ self . clean_session = clean;
82+ self
83+ }
84+
85+ pub fn dynamic_filters ( mut self , dynamic_filters : bool ) -> Self {
86+ self . dynamic_filters = dynamic_filters;
87+ self
88+ }
89+
90+ pub fn build ( self ) -> Result < ( LinkTx , LinkRx , Notification ) , LinkError > {
91+ // Connect to router
92+ // Local connections to the router shall have access to all subscriptions
93+ let connection = Connection :: new (
94+ self . tenant_id ,
95+ self . client_id . to_owned ( ) ,
96+ self . clean_session ,
97+ self . last_will ,
98+ self . dynamic_filters ,
99+ self . topic_alias_max ,
100+ ) ;
101+ let incoming = Incoming :: new ( connection. client_id . to_owned ( ) ) ;
102+ let ( outgoing, link_rx) = Outgoing :: new ( connection. client_id . to_owned ( ) ) ;
103+ let outgoing_data_buffer = outgoing. buffer ( ) ;
104+ let incoming_data_buffer = incoming. buffer ( ) ;
105+
106+ let event = Event :: Connect {
107+ connection,
108+ incoming,
109+ outgoing,
110+ } ;
111+
112+ self . router_tx . send ( ( 0 , event) ) ?;
113+
114+ link_rx. recv ( ) ?;
115+ let notification = outgoing_data_buffer. lock ( ) . pop_front ( ) . unwrap ( ) ;
116+
117+ // Right now link identifies failure with dropped rx in router,
118+ // which is probably ok. We need this here to get id assigned by router
119+ let id = match notification {
120+ Notification :: DeviceAck ( Ack :: ConnAck ( id, ..) ) => id,
121+ _message => return Err ( LinkError :: NotConnectionAck ) ,
122+ } ;
123+
124+ let tx = LinkTx :: new ( id, self . router_tx . clone ( ) , incoming_data_buffer) ;
125+ let rx = LinkRx :: new ( id, self . router_tx , link_rx, outgoing_data_buffer) ;
126+ Ok ( ( tx, rx, notification) )
127+ }
128+ }
129+
38130pub struct Link ;
39131
132+ #[ deprecated = "Link will be removed soon, please consider using LinkBuilder" ]
40133impl Link {
41134 #[ allow( clippy:: type_complexity) ]
42135 fn prepare (
0 commit comments