@@ -6,70 +6,184 @@ mod unix;
66#[ cfg( windows) ]
77mod windows;
88
9+ use futures:: SinkExt ;
10+ use tokio:: {
11+ io,
12+ io:: { AsyncRead , AsyncWrite } ,
13+ } ;
14+ use tokio_stream:: { Stream , StreamExt } ;
15+ use tokio_util:: codec:: Framed ;
16+
917#[ cfg( all( unix, feature = "service" ) ) ]
1018pub use self :: unix:: service;
1119#[ cfg( unix) ]
12- pub use self :: unix:: { DaemonClient , DaemonListener , DaemonLock , DaemonServer } ;
20+ pub use self :: unix:: { DaemonCxAttachment , DaemonLock , NativeDaemonConnector , NativeDaemonListener } ;
1321#[ cfg( all( windows, feature = "service" ) ) ]
1422pub use self :: windows:: service;
1523#[ cfg( windows) ]
16- pub use self :: windows:: { DaemonClient , DaemonListener , DaemonLock , DaemonServer } ;
17- use crate :: { DEFAULT_DAEMON_NAME , DaemonError , DaemonScope } ;
24+ pub use self :: windows:: {
25+ DaemonCxAttachment , DaemonLock , NativeDaemonConnector , NativeDaemonListener ,
26+ } ;
27+ use crate :: {
28+ DaemonCx , DaemonError ,
29+ codec:: { ClientCodec , ServerCodec } ,
30+ cx:: DaemonCxPayload ,
31+ msg:: { Request , Response } ,
32+ } ;
1833
1934#[ cfg( not( any( windows, unix) ) ) ]
2035compile_error ! ( "daemon is not supported on this platform" ) ;
2136
2237impl DaemonLock {
2338 pub fn system ( ) -> Result < Self , DaemonError > {
24- Self :: try_acquire_lock ( DEFAULT_DAEMON_NAME , DaemonScope :: System )
39+ Self :: try_acquire_lock ( DaemonCxPayload :: system ( ) )
2540 }
2641
2742 pub fn user ( ) -> Result < Self , DaemonError > {
28- Self :: try_acquire_lock ( DEFAULT_DAEMON_NAME , DaemonScope :: User )
43+ Self :: try_acquire_lock ( DaemonCxPayload :: user ( ) )
44+ }
45+ }
46+
47+ pub trait DaemonListener : Stream < Item = io:: Result < DaemonServer < Self :: Stream > > > + Sized {
48+ type Stream : AsyncRead + AsyncWrite + Unpin ;
49+ fn new ( ) -> io:: Result < Self > {
50+ Self :: from_context ( DaemonCx :: get ( ) )
51+ }
52+ fn from_context ( cx : DaemonCx ) -> io:: Result < Self > ;
53+ }
54+
55+ pub type NativeDaemonServer = DaemonServer < <NativeDaemonListener as DaemonListener >:: Stream > ;
56+
57+ #[ derive( Debug ) ]
58+ pub struct DaemonServer < T > {
59+ cx : DaemonCx ,
60+ framed : Framed < T , ServerCodec > ,
61+ }
62+ impl < T : AsyncRead + AsyncWrite + Unpin > DaemonServer < T > {
63+ pub fn new ( cx : DaemonCx , inner : T ) -> Self {
64+ Self {
65+ cx,
66+ framed : Framed :: new ( inner, ServerCodec :: new ( ) ) ,
67+ }
68+ }
69+ pub fn context ( & self ) -> DaemonCx {
70+ self . cx
71+ }
72+ pub async fn recv ( & mut self ) -> io:: Result < Request > {
73+ self . framed . next ( ) . await . unwrap_or_else ( || {
74+ Err ( io:: Error :: new (
75+ io:: ErrorKind :: UnexpectedEof ,
76+ "connection closed" ,
77+ ) )
78+ } )
79+ }
80+ pub async fn send ( & mut self , response : Response ) -> io:: Result < ( ) > {
81+ self . framed . send ( response) . await
82+ }
83+ }
84+
85+ pub struct DaemonClient < C : DaemonConnector = NativeDaemonConnector > {
86+ cx : DaemonCx ,
87+ framed : Framed < C :: Stream , ClientCodec > ,
88+ }
89+ impl < C : DaemonConnector + Default > DaemonClient < C > {
90+ pub async fn connect ( ) -> Result < Self , DaemonError > {
91+ let cx = DaemonCx :: get ( ) ;
92+ let connector = C :: default ( ) ;
93+ Self :: connect_with ( cx, connector) . await
94+ }
95+ }
96+ impl < C : DaemonConnector > DaemonClient < C > {
97+ pub async fn connect_with ( cx : DaemonCx , mut connector : C ) -> Result < Self , DaemonError > {
98+ let stream = connector. connect ( cx) . await ?;
99+ let framed = Framed :: new ( stream, ClientCodec :: new ( ) ) ;
100+ Ok ( Self { cx, framed } )
101+ }
102+ pub fn context ( & self ) -> DaemonCx {
103+ self . cx
104+ }
105+ pub async fn send ( & mut self , request : Request ) -> io:: Result < Response > {
106+ self . framed . send ( request) . await ?;
107+ self . framed . next ( ) . await . unwrap_or_else ( || {
108+ Err ( io:: Error :: new (
109+ io:: ErrorKind :: UnexpectedEof ,
110+ "connection closed" ,
111+ ) )
112+ } )
113+ }
114+ }
115+
116+ pub trait DaemonConnector {
117+ type Stream : AsyncRead + AsyncWrite + Unpin ;
118+ type Future : Future < Output = Result < Self :: Stream , DaemonError > > ;
119+ fn connect ( & mut self , cx : DaemonCx ) -> Self :: Future ;
120+ }
121+ impl < T , F , Fut > DaemonConnector for F
122+ where
123+ T : AsyncRead + AsyncWrite + Unpin ,
124+ F : FnMut ( DaemonCx ) -> Fut ,
125+ Fut : Future < Output = Result < T , DaemonError > > ,
126+ {
127+ type Stream = T ;
128+ type Future = Fut ;
129+ fn connect ( & mut self , cx : DaemonCx ) -> Self :: Future {
130+ self ( cx)
29131 }
30132}
31133
32134#[ cfg( test) ]
33135mod tests {
136+ use std:: sync:: OnceLock ;
137+
138+ use color_eyre:: Result ;
139+ use ntest:: timeout;
34140 use pretty_assertions:: { assert_eq, assert_matches} ;
35- use tokio:: time:: timeout;
36141
37142 use crate :: {
38- DaemonId ,
143+ DaemonCx ,
39144 msg:: { Request , Response } ,
40145 test_utils:: * ,
41146 } ;
42147
148+ static FIXTURE : OnceLock < ( ) > = OnceLock :: new ( ) ;
149+
150+ fn try_init ( ) -> Result < ( ) > {
151+ FIXTURE . get_or_try_init ( color_eyre:: install) . copied ( )
152+ }
153+
43154 #[ tokio:: test]
155+ #[ timeout( 500 ) ]
44156 async fn connect_multiple_clients ( ) {
45157 const CLIENTS : usize = 10 ;
46158
47- let id = DaemonId :: random ( ) ;
48- let listener = spawn_listener ( & id , CLIENTS ) . await ;
49- let clients = spawn_clients ( & id , CLIENTS , None ) . await ;
159+ let cx = DaemonCx :: random_in_user_scope ( ) ;
160+ let listener = spawn_listener ( cx , CLIENTS ) . await ;
161+ let clients = spawn_clients ( cx , CLIENTS , 0 ) . await ;
50162
51163 assert_no_err ! ( clients) ;
52164 assert_join_ok ! ( Some ( listener) ) ;
53165 }
54166
55167 #[ tokio:: test]
56- async fn server_echoes_message ( ) {
57- let id = DaemonId :: random ( ) ;
58- let listener = spawn_echo_listener ( & id, 1 ) . await ;
59- let mut client = spawn_clients ( & id, 1 , None ) . await . remove ( 0 ) . unwrap ( ) ;
168+ async fn server_echoes_message ( ) -> Result < ( ) > {
169+ try_init ( ) ?;
170+ let cx = DaemonCx :: random_in_user_scope ( ) ;
171+ let listener = tokio:: spawn ( spawn_echo_listener ( cx, 1 , 500 , 500 ) . with_timeout ( 500 ) ) ;
172+ let mut client = spawn_clients ( cx, 1 , 0 ) . await . remove ( 0 ) ?;
60173
61174 let request = Request {
62175 kind : 42 ,
63176 data : "ping" . to_owned ( ) ,
64177 } ;
65- let response = timeout ( millis ( 1500 ) , client. send ( request. clone ( ) ) ) . await ;
178+ let response = client. send ( request. clone ( ) ) . with_timeout ( 500 ) . await ;
66179
67180 assert_matches ! ( response, Ok ( Ok ( Response :: Echo ( _) ) ) ) ;
68181 let Ok ( Ok ( Response :: Echo ( response) ) ) = response else {
69182 unreachable ! ( ) ;
70183 } ;
71184 assert_eq ! ( request, response) ;
72185 let listener = assert_join_ok ! ( Some ( listener) ) ;
73- let _ = assert_join_ok ! ( flatten!( listener) ) ;
186+ // let _ = assert_join_ok!(flatten!(listener));
187+ Ok ( ( ) )
74188 }
75189}
0 commit comments