@@ -2,7 +2,7 @@ use duva::domains::caches::cache_manager::IndexedValueCodec;
22use duva:: domains:: replications:: LogEntry ;
33use duva:: prelude:: BinBytes ;
44use duva:: prelude:: anyhow:: { self , Context } ;
5- use duva:: presentation:: clients:: request:: NonMutatingAction ;
5+ use duva:: presentation:: clients:: request:: { NonMutatingAction , ServerResponse } ;
66use duva:: {
77 domains:: query_io:: QueryIO , prelude:: tokio:: sync:: oneshot,
88 presentation:: clients:: request:: ClientAction ,
@@ -22,16 +22,21 @@ impl CommandQueue {
2222 self . queue . pop_front ( )
2323 }
2424
25- pub ( crate ) fn finalize_or_requeue ( & mut self , query_io : QueryIO , mut context : InputContext ) {
25+ pub ( crate ) fn finalize_or_requeue (
26+ & mut self ,
27+ query_io : ServerResponse ,
28+ mut context : InputContext ,
29+ ) {
2630 context. results . push ( query_io) ;
2731
2832 if context. results . len ( ) != context. expected_result_cnt {
2933 self . push ( context) ;
3034 return ;
3135 }
3236
33- let result =
34- context. get_result ( ) . unwrap_or_else ( |err| QueryIO :: Err ( BinBytes :: new ( err. to_string ( ) ) ) ) ;
37+ let result = context
38+ . get_result ( )
39+ . unwrap_or_else ( |err| ServerResponse :: Err { reason : err. to_string ( ) , request_id : 0 } ) ;
3540 context. callback ( result) ;
3641 }
3742}
@@ -47,19 +52,19 @@ pub fn separate_command_and_args(args: Vec<&str>) -> (&str, Vec<&str>) {
4752#[ derive( Debug ) ]
4853pub struct InputContext {
4954 pub ( crate ) client_action : ClientAction ,
50- pub ( crate ) callback : oneshot:: Sender < ( ClientAction , QueryIO ) > ,
51- pub ( crate ) results : Vec < QueryIO > ,
55+ pub ( crate ) callback : oneshot:: Sender < ( ClientAction , ServerResponse ) > ,
56+ pub ( crate ) results : Vec < ServerResponse > ,
5257 pub ( crate ) expected_result_cnt : usize ,
5358}
5459impl InputContext {
5560 pub fn new (
5661 client_action : ClientAction ,
57- callback : oneshot:: Sender < ( ClientAction , QueryIO ) > ,
62+ callback : oneshot:: Sender < ( ClientAction , ServerResponse ) > ,
5863 ) -> Self {
5964 Self { client_action, callback, results : Vec :: new ( ) , expected_result_cnt : 0 }
6065 }
6166
62- pub ( crate ) fn callback ( self , query_io : QueryIO ) {
67+ pub ( crate ) fn callback ( self , query_io : ServerResponse ) {
6368 let action_debug = format ! ( "{:?}" , self . client_action) ;
6469 self . callback . send ( ( self . client_action , query_io) ) . unwrap_or_else ( |_| {
6570 // Log callback failure for debugging
@@ -70,51 +75,67 @@ impl InputContext {
7075 } ) ;
7176 }
7277
73- pub ( crate ) fn get_result ( & mut self ) -> anyhow:: Result < QueryIO > {
78+ pub ( crate ) fn get_result ( & mut self ) -> anyhow:: Result < ServerResponse > {
7479 use NonMutatingAction :: * ;
7580 let res = std:: mem:: take ( & mut self . results ) ;
81+ let mut highest_req_id = 0 ; // TODO for now, set request id to the highest one
82+ let mut iterator = res. into_iter ( ) ;
7683
7784 match self . client_action {
7885 ClientAction :: NonMutating ( Keys { pattern : _ } | MGet { keys : _ } ) => {
79- let mut init = QueryIO :: Array ( Vec :: with_capacity ( res. len ( ) ) ) ;
80- for item in res {
81- init = init. merge ( item) ?;
86+ let mut init = QueryIO :: Array ( Vec :: with_capacity ( iterator. len ( ) ) ) ;
87+
88+ while let Some ( ServerResponse :: ReadRes { res, request_id } ) = iterator. next ( ) {
89+ init = init. merge ( res) ?;
90+ highest_req_id = highest_req_id. max ( request_id) ;
8291 }
83- Ok ( init)
92+ Ok ( ServerResponse :: ReadRes { res : init, request_id : highest_req_id } )
8493 } ,
94+
8595 ClientAction :: NonMutating ( Exists { keys : _ } ) => {
8696 let mut count = 0 ;
87- for result in res {
88- let QueryIO :: SimpleString ( byte) = result else {
89- return Err ( anyhow:: anyhow!( "Expected SimpleString result" ) ) ;
90- } ;
97+
98+ while let Some ( ServerResponse :: ReadRes {
99+ res : QueryIO :: BulkString ( byte) ,
100+ request_id,
101+ } ) = iterator. next ( )
102+ {
91103 let num = String :: from_utf8 ( byte. to_vec ( ) )
92104 . context ( "Failed to convert byte to string" ) ?;
93105 let num = num. parse :: < u64 > ( ) . context ( "Failed to parse string to u64" ) ?;
94106
95107 count += num;
108+ highest_req_id = highest_req_id. max ( request_id) ;
96109 }
97- Ok ( QueryIO :: SimpleString ( BinBytes :: new ( count. to_string ( ) ) ) )
110+
111+ Ok ( ServerResponse :: ReadRes {
112+ res : QueryIO :: BulkString ( BinBytes :: new ( count. to_string ( ) ) ) ,
113+ request_id : highest_req_id,
114+ } )
98115 } ,
99116 ClientAction :: Mutating ( LogEntry :: Delete { keys : _ } ) => {
100117 let mut count = 0 ;
101- for result in res {
102- let QueryIO :: SimpleString ( value) = result else {
103- return Err ( anyhow:: anyhow!( "Expected SimpleString result" ) ) ;
104- } ;
118+
119+ while let Some ( ServerResponse :: WriteRes {
120+ res : QueryIO :: BulkString ( value) ,
121+ request_id,
122+ ..
123+ } ) = iterator. next ( )
124+ {
105125 let decoded_value =
106126 IndexedValueCodec :: decode_value ( String :: from_utf8_lossy ( & value) ) . unwrap ( ) ;
107127
108128 count += decoded_value;
129+ highest_req_id = highest_req_id. max ( request_id) ;
109130 }
110- Ok ( QueryIO :: SimpleString ( BinBytes :: new ( count. to_string ( ) ) ) )
111- } ,
112- _ => {
113- if res. len ( ) != 1 {
114- return Err ( anyhow:: anyhow!( "Expected exactly one result" ) ) ;
115- }
116- Ok ( res[ 0 ] . clone ( ) )
131+
132+ Ok ( ServerResponse :: WriteRes {
133+ res : QueryIO :: BulkString ( BinBytes :: new ( count. to_string ( ) ) ) ,
134+ log_index : 0 , // TODO
135+ request_id : highest_req_id,
136+ } )
117137 } ,
138+ _ => iterator. next ( ) . ok_or ( anyhow:: anyhow!( "Expected exactly one result" ) ) ,
118139 }
119140 }
120141}
0 commit comments