Fixed small issues

Author: drewhk

protocol.md

@@ -19,6 +19,8 @@ The `type` field indicates that this is an API request, the `serviceId` field in
 will bear the `requestId` of 652. Finally, the `payload` field contains the JSON formatted message that is to be
 parsed by the particular service, in this case `getCustomerIds`.
 
+If the `requestId` was already used by a previous request, then that request is implicitly cancelled.
+
 On a successful call, the backend will respond with a stream of messages:
 
 ```json
@@ -62,7 +64,7 @@ In the case the client sends a request to a non-existing service, for example:
 {"type":"request","serviceId":"getCustomerIdsWrong","requestId":652,"payload":{ ... }}
 ```
 
-The server responds with an error of `unknonwnEndpoint`:
+The server responds with an error of `unknownEndpoint`:
 
 ```json
 {"type":"error","requestId":652,"kind":{"type":"unknownEndpoint","endpoint":"getCustomerIdsWrong"}}

src/lib.rs

@@ -97,7 +97,6 @@ pub fn serve(services: Vec<BoxedService>) -> impl Filter<Extract = (impl warp::R
     })
 }
 
-// TODO: Cancel previous requests
 fn client_connected(
     ws: WebSocket,
     services: Arc<BTreeMap<String, BoxedService>>,
@@ -137,7 +136,11 @@ fn client_connected(
                             if let Some(srv) = services.get(body.service_id) {
                                 // Set up cancellation signal
                                 let canceled = Arc::new(AtomicBool::new(false));
-                                active_responses.insert(body.request_id, canceled.clone());
+
+                                if let Some(previous) = active_responses.insert(body.request_id, canceled.clone()) {
+                                    previous.store(true, Ordering::SeqCst);
+                                };
+
                                 executor
                                     .spawn(serve_request(
                                         canceled,
@@ -367,7 +370,7 @@ mod tests {
                     completion = Some(env.clone());
                     false
                 }
-            }) // if let Outgoing::Next { .. } = env { true } else { false })
+            })
             .filter_map(|env| {
                 if let Outgoing::Next { payload, .. } = env {
                     Some(serde_json::from_value::<Resp>(payload).expect("Could not deserialize response"))