Parse metadata in EventSource

components/script/dom/eventsource.rs

@@ -6,7 +6,9 @@ use dom::bindings::cell::DOMRefCell;
 use dom::bindings::codegen::Bindings::EventHandlerBinding::EventHandlerNonNull;
 use dom::bindings::codegen::Bindings::EventSourceBinding::{EventSourceInit, EventSourceMethods, Wrap};
 use dom::bindings::error::{Error, Fallible};
+use dom::bindings::inheritance::Castable;
 use dom::bindings::js::Root;
+use dom::bindings::refcounted::Trusted;
 use dom::bindings::reflector::reflect_dom_object;
 use dom::bindings::str::DOMString;
 use dom::eventtarget::EventTarget;
@@ -14,19 +16,22 @@ use dom::globalscope::GlobalScope;
 use hyper::header::{Accept, qitem};
 use ipc_channel::ipc;
 use ipc_channel::router::ROUTER;
+use mime::{Mime, TopLevel, SubLevel};
 use net_traits::{CoreResourceMsg, FetchMetadata, FetchResponseListener, NetworkError};
 use net_traits::request::{CacheMode, CorsSettings, CredentialsMode};
 use net_traits::request::{RequestInit, RequestMode};
 use network_listener::{NetworkListener, PreInvoke};
+use script_thread::{Runnable, RunnableWrapper};
 use std::cell::Cell;
 use std::sync::{Arc, Mutex};
+use task_source::TaskSource;
+use task_source::networking::NetworkingTaskSource;
 use url::Url;
 
 #[derive(JSTraceable, PartialEq, Copy, Clone, Debug, HeapSizeOf)]
 /// https://html.spec.whatwg.org/multipage/#dom-eventsource-readystate
 enum ReadyState {
     Connecting = 0,
-    #[allow(dead_code)]
     Open = 1,
     Closed = 2
 }
@@ -36,12 +41,32 @@ pub struct EventSource {
     eventtarget: EventTarget,
     url: DOMRefCell<Option<Url>>,
     request: DOMRefCell<Option<RequestInit>>,
-    ready_state: Cell<EventSourceReadyState>,
+    ready_state: Cell<ReadyState>,
     with_credentials: bool,
     last_event_id: DOMRefCell<DOMString>
 }
 
-struct EventSourceContext;
+struct EventSourceContext {
+    event_source: Trusted<EventSource>,
+    networking_task_source: NetworkingTaskSource,
+    wrapper: RunnableWrapper
+}
+
+impl EventSourceContext {
+    fn announce_the_connection(&self) {
+        let runnable = box AnnounceConnectionRunnable {
+            event_source: self.event_source.clone()
+        };
+        let _ = self.networking_task_source.queue_with_wrapper(runnable, &self.wrapper);
+    }
+
+    fn fail_the_connection(&self) {
+        let runnable = box FailConnectionRunnable {
+            event_source: self.event_source.clone()
+        };
+        let _ = self.networking_task_source.queue_with_wrapper(runnable, &self.wrapper);
+    }
+}
 
 impl FetchResponseListener for EventSourceContext {
     fn process_request_body(&mut self) {
@@ -52,8 +77,27 @@ impl FetchResponseListener for EventSourceContext {
         // TODO
     }
 
-    fn process_response(&mut self, _metadata: Result<FetchMetadata, NetworkError>) {
-        // TODO
+    fn process_response(&mut self, metadata: Result<FetchMetadata, NetworkError>) {
+        match metadata {
+            Ok(fm) => {
+                let meta = match fm {
+                    FetchMetadata::Unfiltered(m) => m,
+                    FetchMetadata::Filtered { unsafe_, .. } => unsafe_
+                };
+                match meta.content_type {
+                    None => self.fail_the_connection(),
+                    Some(ct) => match ct.into_inner().0 {
+                        Mime(TopLevel::Text, SubLevel::EventStream, _) =>
+                            self.announce_the_connection(),
+                        _ => self.fail_the_connection()
+                    }
+                }
+            }
+            Err(_) => {
+                // FIXME: Fail the connection for now, but it should really attempt to re-establish
+                self.fail_the_connection();
+            }
+        }
     }
 
     fn process_response_chunk(&mut self, mut _chunk: Vec<u8>) {
@@ -129,7 +173,11 @@ impl EventSource {
         // Step 12
         *ev.request.borrow_mut() = Some(request.clone());
         // Step 14
-        let context = EventSourceContext;
+        let context = EventSourceContext {
+            event_source: Trusted::new(&ev),
+            networking_task_source: global.networking_task_source(),
+            wrapper: global.get_runnable_wrapper()
+        };
         let listener = NetworkListener {
             context: Arc::new(Mutex::new(context)),
             task_source: global.networking_task_source(),
@@ -176,3 +224,37 @@ impl EventSourceMethods for EventSource {
         // TODO: Terminate ongoing fetch
     }
 }
+
+pub struct AnnounceConnectionRunnable {
+    event_source: Trusted<EventSource>,
+}
+
+impl Runnable for AnnounceConnectionRunnable {
+    fn name(&self) -> &'static str { "EventSource AnnounceConnectionRunnable" }
+
+    // https://html.spec.whatwg.org/multipage/#announce-the-connection
+    fn handler(self: Box<AnnounceConnectionRunnable>) {
+        let event_source = self.event_source.root();
+        if event_source.ready_state.get() != ReadyState::Closed {
+            event_source.ready_state.set(ReadyState::Open);
+            event_source.upcast::<EventTarget>().fire_event(atom!("open"));
+        }
+    }
+}
+
+pub struct FailConnectionRunnable {
+    event_source: Trusted<EventSource>,
+}
+
+impl Runnable for FailConnectionRunnable {
+    fn name(&self) -> &'static str { "EventSource FailConnectionRunnable" }
+
+    // https://html.spec.whatwg.org/multipage/#fail-the-connection
+    fn handler(self: Box<FailConnectionRunnable>) {
+        let event_source = self.event_source.root();
+        if event_source.ready_state.get() != ReadyState::Closed {
+            event_source.ready_state.set(ReadyState::Closed);
+            event_source.upcast::<EventTarget>().fire_event(atom!("error"));
+        }
+    }
+}