Google App Engine ndb async support

README.md

@@ -35,6 +35,22 @@ Credentials can also be set in a per-instance basis:
 
     p = pusher.Pusher(app_id='your-pusher-app-id', key='your-pusher-key', secret='your-pusher-secret')
 
+## Custom JSON
+
+If you need custom JSON serialization for handling dates or custom data types, configure pusher to use your own json module. Pusher expects the module to contain a `dumps` function. In the example, the custom JSON module is named `custom_json_module`:
+
+    pusher.json = custom_json_module
+
+If you're using a custom encoder class, create a stub class with `dumps` method masquerading as a json module, locking the cls input argument using partial:
+
+    from functools import partial
+    pusher.json = type('CustomJson', (object,), dict(dumps=staticmethod(partial(json.dumps, cls=json.JSONEncoder))))
+
+Or following the same pattern, setting a default encode method named `default_encode_func`:
+
+    from functools import partial
+    pusher.json = type('CustomJson', (object,), dict(dumps=staticmethod(partial(json.dumps, default=default_encode_func))))
+
 ## Heroku
 
 If you're using Pusher as a Heroku add-on, you can just get the config informat
@@ -55,7 +71,11 @@ To force the module to use AppEngine's urlfetch, do the following on setup:
 
     pusher.channel_type = pusher.GoogleAppEngineChannel
 
-I haven't been able to test this though. Can somebody confirm it works? Thanks! `:-)`
+## Google AppEngine NDB
+
+A channel that uses the GAE NDB (introduced in GAE 1.6.3) and provides async methods that return a future, `trigger_async` and `send_request_async`.
+
+    pusher.channel_type = pusher.GaeNdbChannel
 
 ## Running the tests
 

pusher/__init__.py

@@ -132,15 +132,51 @@ def authentication_string(self, socket_id, custom_string=None):
 class GoogleAppEngineChannel(Channel):
     def send_request(self, query_string, data_string):
         from google.appengine.api import urlfetch
-        absolute_url = 'http://%s/%s?%s' % (self.pusher.host, self.path, query_string)
+        absolute_url = 'http://%s%s?%s' % (self.pusher.host, self.path, query_string)
         response = urlfetch.fetch(
             url=absolute_url,
             payload=data_string,
             method=urlfetch.POST,
             headers={'Content-Type': 'application/json'}
         )
         return response.status_code
-
+
+# App Engine NDB channel, outer try import/except as it uses decorator
+try:
+    from google.appengine.ext import ndb
+
+    class GaeNdbChannel(GoogleAppEngineChannel):
+        @ndb.tasklet
+        def trigger_async(self, event, data={}, socket_id=None):
+            """Async trigger that in turn calls send_request_async"""
+            json_data = json.dumps(data)
+            status = yield self.send_request_async(self.signed_query(event, json_data, socket_id), json_data)
+            if status == 202:
+                raise ndb.Return(True)
+            elif status == 401:
+                raise AuthenticationError
+            elif status == 404:
+                raise NotFoundError
+            else:
+                raise Exception("Unexpected return status %s" % status)
+
+        @ndb.tasklet
+        def send_request_async(self, query_string, data_string):
+            """Send request and yield while waiting for future result"""
+            ctx = ndb.get_context()
+            secure = 's' if self.pusher.port == 443 else ''
+            absolute_url = 'http%s://%s%s?%s' % (secure, self.pusher.host, self.path, query_string)
+            result = yield ctx.urlfetch(
+                url=absolute_url,
+                payload=data_string,
+                method='POST',
+                headers={'Content-Type': 'application/json'},
+                validate_certificate=bool(secure),
+                )
+            raise ndb.Return(result.status_code)
+except ImportError:
+    pass
+
 class TornadoChannel(Channel):
     def trigger(self, event, data={}, socket_id=None, callback=None):
         self.callback = callback