cloudevents · evankanderson · Jan 12, 2019 · Jan 12, 2019 · Jan 12, 2019
diff --git a/cloudevents/sdk/converters/structured.py b/cloudevents/sdk/converters/structured.py
@@ -23,10 +23,19 @@ class JSONHTTPCloudEventConverter(base.Converter):
 
     TYPE = "structured"
 
+    MIME_TYPE = "application/cloudevents+json"
+
     def read(self, event: event_base.BaseEvent,
              headers: dict,
              body: typing.IO,
              data_unmarshaller: typing.Callable) -> event_base.BaseEvent:
+        # Note: this is fragile for true dictionaries which don't implement
+        # case-insensitive header mappings. HTTP/1.1 specifies that headers
+        #  are case insensitive, so this usually affects tests.
+        if not headers.get("Content-Type", "").startswith(self.MIME_TYPE):
+            raise exceptions.UnsupportedEvent(
+                "Structured mode must be {0}, not {1}".format(
+                    self.MIME_TYPE, headers.get("content-type")))
         event.UnmarshalJSON(body, data_unmarshaller)
         return event
 

diff --git a/cloudevents/sdk/event/base.py b/cloudevents/sdk/event/base.py
@@ -129,31 +129,33 @@ def UnmarshalJSON(self, b: typing.IO,
 
     def UnmarshalBinary(self, headers: dict, body: typing.IO,
                         data_unmarshaller: typing.Callable):
-        props = self.Properties(with_nullable=True)
-        exts = props.get("extensions")
-        for key in props:
-            formatted_key = "ce-{0}".format(key)
-            if key != "extensions":
-                self.Set(key, headers.get("ce-{0}".format(key)))
-                if formatted_key in headers:
-                    del headers[formatted_key]
-
-        # rest of headers suppose to an extension?
-        exts.update(**headers)
-        self.Set("extensions", exts)
+        BINARY_MAPPING = {
+            'content-type': 'contenttype',
+            # TODO(someone): add Distributed Tracing. It's not clear if this
+            # is one extension or two.
+            # https://github.com/cloudevents/spec/blob/master/extensions/distributed-tracing.md
+        }
+        for header, value in headers.items():
+            header = header.lower()
+            if header in BINARY_MAPPING:
+                self.Set(BINARY_MAPPING[header], value)
+            elif header.startswith("ce-"):
+                self.Set(header[3:], value)
+
         self.Set("data", data_unmarshaller(body))
 
     def MarshalBinary(self) -> (dict, object):
         headers = {}
+        if self.ContentType():
+            headers["content-type"] = self.ContentType()
         props = self.Properties()
         for key, value in props.items():
-            if key not in ["data", "extensions"]:
+            if key not in ["data", "extensions", "contenttype"]:
                 if value is not None:
                     headers["ce-{0}".format(key)] = value
 
-        exts = props.get("extensions")
-        if len(exts) > 0:
-            headers.update(**exts)
+        for key, value in props.get("extensions"):
+            headers["ce-{0}".format(key)] = value
 
         data, _ = self.Get("data")
         return headers, data
diff --git a/cloudevents/sdk/marshaller.py b/cloudevents/sdk/marshaller.py
@@ -56,7 +56,12 @@ def FromRequest(self, event: event_base.BaseEvent,
         :rtype: event_base.BaseEvent
         """
         for _, cnvrtr in self.__converters.items():
-            return cnvrtr.read(event, headers, body, data_unmarshaller)
+            try:
+                return cnvrtr.read(event, headers, body, data_unmarshaller)
+            except exceptions.UnsupportedEvent:
+                continue
+        raise exceptions.UnsupportedEvent(
+            "No registered marshaller in {0}".format(self.__converters))
 
     def ToRequest(self, event: event_base.BaseEvent,
                   converter_type: str,

diff --git a/cloudevents/tests/test_event_from_request_converter.py b/cloudevents/tests/test_event_from_request_converter.py
@@ -90,7 +90,7 @@ def test_structured_converter_v01():
     assert event.Get("id") == (data.ce_id, True)
 
 
-def test_default_http_marshaller():
+def test_default_http_marshaller_with_structured():
     m = marshaller.NewDefaultHTTPMarshaller()
 
     event = m.FromRequest(
@@ -102,3 +102,18 @@ def test_default_http_marshaller():
     assert event is not None
     assert event.Get("type") == (data.ce_type, True)
     assert event.Get("id") == (data.ce_id, True)
+
+
+def test_default_http_marshaller_with_binary():
+    m = marshaller.NewDefaultHTTPMarshaller()
+
+    event = m.FromRequest(
+        v02.Event(),
+        data.headers,
+        io.StringIO(ujson.dumps(data.body)),
+        ujson.load
+    )
+    assert event is not None
+    assert event.Get("type") == (data.ce_type, True)
+    assert event.Get("data") == (data.body, True)
+    assert event.Get("id") == (data.ce_id, True)
diff --git a/cloudevents/tests/test_event_pipeline.py b/cloudevents/tests/test_event_pipeline.py
@@ -42,7 +42,7 @@ def test_event_pipeline_upstream():
     assert "ce-source" in new_headers
     assert "ce-id" in new_headers
     assert "ce-time" in new_headers
-    assert "ce-contenttype" in new_headers
+    assert "content-type" in new_headers
     assert data.body == body