[WIP] Adding stream listener to stream changes in child nodes #183

the-c0d3r · 2018-07-21T14:17:14Z

Hi,

This PR's original intent is to add stream listener function to fire a callback function when changes in the database is detected. It is a revival of this #50 1-year-old PR and also based on the code contributed by @rizasif and @Aqsa-K in that PR. This PR is mainly to serve 3 purposes.

Update the old PR with a rebase on master
Clean up, refactor & fix lint errors
Write tests

For now, I have completed the first and second of the above. Please advise me on how to properly write the test in PyTest. I tested the code manually by defining callback and manually modifiying data in my firebase demo project. And it seems to be fired everytime I made changes. I read through the existing test suite and I cannot think of a way to write into it yet.

Changes:

lint.sh : Ignore directive for "protected-access"
firebase_admin/sseclient.py : SSE client module
firebase_admin/db.py : Streaming functionality
New API call: database.reference('/xxx').stream(callbackFunc)

Regards

hiranya911 · 2018-07-23T23:57:36Z

@the-c0d3r thanks for putting in the time and effort into fixing this. Some comments and answers:

Please revert the changes to lint.sh. Instead add the directives to the lines in the source files where the lint errors should be ignored.

Rename sseclient.py to _sseclient.py so it will be considered an internal module.

For unit testing, you can mount an adapter on the requests.Session object and have it return some fake events (I think). See how we do something similar in the existing tests:

firebase-admin-python/tests/test_db.py

Lines 128 to 132 in 70d9817

 def instrument(self, ref, payload, status=200): 

 recorder = [] 

 adapter = MockAdapter(payload, status, recorder) 

 ref._client.session.mount(self.test_url, adapter) 

 return recorder

The MockAdapter used in the above example responds to requests made by the SDK with a fake response. You will probably end up writing a similar adapter that responds with a fake stream of events.

…-python into stream_listener

hiranya911 · 2018-07-24T18:24:09Z

firebase_admin/db.py

@@ -155,10 +155,10 @@ def parent(self):

 def build_headers(self, token=None):
 headers = {'content-type' : 'application/json; charset=UTF-8'}
- if not token and self._client._session.credentials:
+ if not token and self._client._session.credentials: # pylint: disable=protected-access


I think self._client.session should work here. session is exposed from the HttpClient:

firebase-admin-python/firebase_admin/_http_client.py

Lines 53 to 55 in c420b4f

@property

def session(self):

return self._session

That way you won't need the linter directives at all.

Oh, I didn't catch that earlier. I have changed it, thanks. I still have yet to write the tests. I'll finish the tests in a few days.

great work @the-c0d3r . I am glad that this is finally progressing. Let me know if any help is required.

the-c0d3r · 2018-07-26T15:09:43Z

Hi, I have added the test for SSEClient code. I'm not sure if I did it correctly or not. I added MockSSEClientAdapter() which is a subclass of MockAdapter. And I captured the actual event raw response and put it in payload. But somehow when I run the test, the MockSSEClient actually sends out Bytes object which is expected, but inside SSEClient code's response.iter_content returns bytes instead of string. I compared it with the my program to actually communicate with the server, and somehow iter_content returns string. Therefore for now, I added a check for the type in SSEClient, to call decode() if the nextchar from response.iter_content is a bytes object.

                nextchar = next(self.resp_iterator)
                if isinstance(nextchar, bytes):
                    nextchar = nextchar.decode()
                self.buf += nextchar

Python will raise TypeError: Can't convert 'bytes' object to str implicitly if executed the test withouth this check. Is there anyway to modify the MockSSEClientAdapter or the parent class, to make it's response.iter_content return string instead of bytes?
But I'm not sure about this check yet so I haven't committed this change. Please comment.

hiranya911 · 2018-07-26T18:48:45Z

Hi @the-c0d3r. I think this is due to the way iter_content() is implemented in requests:

http://docs.python-requests.org/en/master/_modules/requests/models/#Response.iter_content

There are couple of things you can try out.

Try setting resp.raw.stream = True before returning from send().
Try setting an encoding on the mock response as it is expected here.

One of these is likely to fix the issue.

Update:

On a second look, resp.raw.stream is expected to be a function, not a boolean. So option 2 is probably the only viable option.

the-c0d3r · 2018-07-27T03:23:52Z

A quick test shows that the second fix you suggested works. I'll patch it in and commit that later today. And I also found the CI build failed for python3's syntax super() call in python2 environment. I'll change it back to support both python3 and python2.

Are there any additional tasks left to be completed for this PR to be merged?
PS: Does Stream class needs to have testing? I noticed that the SSE client's testing code more or less does the same thing as the Stream class.

hiranya911 · 2018-07-27T17:00:11Z

Thanks @the-c0d3r. This still needs to go through our internal API review process (which I have already initiated). We might get some feedback from that, which will need to be implemented -- but we can do that in steps, after merging this one. I will also take another closer look at all the code here, and post some feedback.

hiranya911

@the-c0d3r A few code review comments to address when you get a chance. But overall this looks pretty good. Thanks again for putting in the effort.

hiranya911 · 2018-07-27T17:05:43Z

firebase_admin/_sseclient.py

@@ -0,0 +1,198 @@
+"""SSEClient module to handle streaming of realtime changes on the database


Please add the Apache license header here.

hiranya911 · 2018-07-27T17:07:30Z

firebase_admin/_sseclient.py

+ # Optional support for passing in a requests.Session()
+ self.session = session
+ # function for building auth header when token expires
+ self.build_headers = build_headers


Do we really need this? The session we get from the db module is an instance of google.auth.transport.requests.AuthorizedSession, which is guaranteed to add the required Authorization header. So this shouldn't be required.

build_header() and all subsequent function arguments removed.

hiranya911 · 2018-07-27T17:09:07Z

firebase_admin/_sseclient.py

+ self.requests_kwargs['headers']['Cache-Control'] = 'no-cache'
+
+ # The 'Accept' header is not required, but explicit > implicit
+ self.requests_kwargs['headers']['Accept'] = 'text/event-stream'


The above can be written as:

headers = self.request_kwargs.get('headers', {}) # add the required values to headers self.request_kwargs['headers'] = headers

hiranya911 · 2018-07-27T17:10:39Z

firebase_admin/_sseclient.py

+ self.should_connect = False
+ self.retry = 0
+ self.resp.close()
+ # self.resp.raw._fp.fp.raw._sock.shutdown(socket.SHUT_RDWR)


Remove the commented out lines?

On a side note, I was testing this branch the other day, and I noticed that calling close() here does not immediately release the underlying socket. As a result the Stream remains active for a while even after calling stream.close(). So perhaps we do need the commented out lines?

I also noticed the delay in calling stream.close(). I was calling it in the python shell, and it takes a few seconds to close and return. I'll test it out a bit more after I followed your other suggestions first.

hiranya911 · 2018-07-27T17:11:10Z

firebase_admin/_sseclient.py

+
+ def _connect(self):
+ """connects to the server using requests"""
+ if self.should_connect:


Should also check for self.running.

hiranya911 · 2018-07-27T17:13:36Z

firebase_admin/db.py


 import firebase_admin
 from firebase_admin import _http_client
 from firebase_admin import _utils
+from firebase_admin._sseclient import SSEClient, KeepAuthSession


As a matter of style, we don't import individual classes and members. Please import the module here as from firebase_admin import _sseclient, and change the code below accordingly.

hiranya911 · 2018-07-27T17:35:13Z

firebase_admin/db.py

+
+ def start_stream(self):
+ """Streaming function for the spawned thread to run"""
+ self.sse = SSEClient(


Can you move this to the start() method above, before the thread is started? Then I think you will be able to drop the sleep() call in close().

hiranya911 · 2018-07-27T17:36:09Z

firebase_admin/db.py

@@ -101,6 +153,23 @@ def parent(self):
 return Reference(client=self._client, segments=self._segments[:-1])
 return None

+ def build_headers(self, token=None):


See if we can remove this altogether (see my comment on _sseclient). If we must keep this, please rename to _build_headers() so it's treated as an internal method.

This build_headers() is tested to be unnecessary just like your comment on _sseclient. So I will remove it from all subsequent code.

hiranya911 · 2018-07-27T17:36:24Z

tests/test_db.py

@@ -523,7 +523,7 @@ def test_range_query(self):
 assert recorder[0].headers['User-Agent'] == db._USER_AGENT


-class TestDatabseInitialization(object):
+class TestDatabaseInitialization(object):


Good catch 👍

hiranya911 · 2018-07-27T17:38:09Z

tests/test_sseclient.py

+ sseclient = self.init_sse()
+ for msg in sseclient:
+ event = json.loads(msg.data)
+ break


Is the break necessary? Shouldn't it exit on its own since the mock response only has one event?

…tEvent

the-c0d3r · 2018-07-28T04:13:18Z

Hi @hiranya911, I believe I have addressed all of your comments with my latest commits, except the part about the closing of the stream. I wrote a simple python code to just setup stream and close it and record the timing. It returned within 100 ms. But somehow when executing the code line by line manually in python shell, it would take over 10 seconds to close the stream (aka time taken for python to return control to user). It will take me some time to debug this since I don't know much about requests module's internals. If you can shed some light on it, that would be great.

hiranya911

Looks pretty good. I just had one comment for you.

We can investigate the socket closing issue later. Once we get the internal approval for this, I can merge it, but I'll probably make a few more changes based on the feedback we get.

hiranya911 · 2018-07-30T17:53:53Z

firebase_admin/_sseclient.py

+ last_id: optional id
+ retry: the interval in ms
+ **kwargs: extra kwargs will be sent to requests.get
+ """
 self.should_connect = True
 self.url = url
 self.last_id = last_id
 self.retry = retry
 self.running = True


Remove this as it gets set to False below.

roboflank · 2018-07-31T05:31:04Z

+1

hiranya911 · 2018-08-02T17:26:55Z

@the-c0d3r the internal review process for this is chugging along. I will have an update for you next week.

hiranya911

LGTM.

Please sync this PR with the latest master when you get a chance.

the-c0d3r · 2018-08-03T01:05:23Z

Hi @hiranya911 thanks for the update. I have synced the PR with the latest master. If there are any further tasks, please let me know.

hiranya911 · 2018-08-07T20:26:23Z

@the-c0d3r I have some feedback. Once they are implemented, I can merge this PR:

Rename the stream() method to listen().
Rename the Stream type to ListenerRegistration.
Remove the stream_id parameter from the listen() method.

Let me know if you have any questions.

…emoved 'stream_id' and added more documentation

the-c0d3r · 2018-08-08T01:34:02Z

Hi, I think I have addressed those feedbacks. Please take a look.

hiranya911 · 2018-08-08T16:47:47Z

Thanks @the-c0d3r. This looks pretty solid. I'll merge it shortly. I'll intend to do a bit of work on top it over the next few days, before release.

hiranya911 · 2018-08-08T18:10:30Z

Upon taking another look, this might have a problem. I don't see how the credentials are passed into the underlying session.

hiranya911 · 2018-08-08T18:55:12Z

Ok, managed to fix the issue. I'll send a separate PR with the fix.

the-c0d3r added 4 commits July 21, 2018 19:05

Added SSEClient library

16cdfcb

Added Streaming functionality to db.py

b6f2eae

Added ignore directive 'protected-access' for db.py in lint.sh

561f4eb

Fixed typo

e6944c1

the-c0d3r and others added 5 commits July 24, 2018 09:07

Merge branch 'master' into stream_listener

93c393a

Renamed file to internal module

4dccb8c

Reverted lint.sh, added ignore directive in db.py

f8ca12c

Merge branch 'stream_listener' of github.com:the-c0d3r/firebase-admin…

f14ebff

…-python into stream_listener

Changed import module name to internal name

6a829d0

hiranya911 reviewed Jul 24, 2018

View reviewed changes

the-c0d3r added 2 commits July 25, 2018 07:31

Fixed pylint protected-access by not calling protected member

4b9a952

Added test_sseclient.py

db96e05

the-c0d3r added 2 commits July 27, 2018 19:46

python2,3 compatibility, fixed encoding issue

0906a79

Added tests for Event() class

ea58b58

hiranya911 suggested changes Jul 27, 2018

View reviewed changes

the-c0d3r added 6 commits July 28, 2018 10:41

removed build_headers() function

104cc4e

Removed Event().dump() test code

8c0c3e7

Changed import to match code style

0186ed9

init sseclient to Stream().start() and removed sleep

37ae857

Added apache license 2.0 header

9a0d4c3

changed import style, removed for loop in event in test_sseclient.Tes…

d9142ad

…tEvent

hiranya911 reviewed Jul 30, 2018

View reviewed changes

Removed self.running from firebase_admin._sseclient

bfb7c69

hiranya911 approved these changes Aug 2, 2018

View reviewed changes

Merge branch 'master' into stream_listener

df05d22

Renamed Stream class to ListenerRegistration, stream() to listen(), r…

76896a5

…emoved 'stream_id' and added more documentation

hiranya911 merged commit 3f1190d into firebase:master Aug 8, 2018

This was referenced Aug 8, 2018

Fixed and refactored the db.Reference.listen() API #190

Merged

Adding stream listeners to sniff change in child #50

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[WIP] Adding stream listener to stream changes in child nodes #183

[WIP] Adding stream listener to stream changes in child nodes #183

the-c0d3r commented Jul 21, 2018 •

edited

hiranya911 commented Jul 23, 2018 •

edited

hiranya911 Jul 24, 2018

the-c0d3r Jul 24, 2018

Aqsa-K Jul 25, 2018

the-c0d3r commented Jul 26, 2018

hiranya911 commented Jul 26, 2018 •

edited

the-c0d3r commented Jul 27, 2018

hiranya911 commented Jul 27, 2018

hiranya911 left a comment

hiranya911 Jul 27, 2018

hiranya911 Jul 27, 2018

the-c0d3r Jul 28, 2018

hiranya911 Jul 27, 2018

hiranya911 Jul 27, 2018

the-c0d3r Jul 28, 2018 •

edited

hiranya911 Jul 27, 2018

hiranya911 Jul 27, 2018

hiranya911 Jul 27, 2018

hiranya911 Jul 27, 2018

the-c0d3r Jul 28, 2018

hiranya911 Jul 27, 2018

hiranya911 Jul 27, 2018

the-c0d3r commented Jul 28, 2018

hiranya911 left a comment

hiranya911 Jul 30, 2018

roboflank commented Jul 31, 2018

hiranya911 commented Aug 2, 2018

hiranya911 left a comment

the-c0d3r commented Aug 3, 2018

hiranya911 commented Aug 7, 2018

the-c0d3r commented Aug 8, 2018

hiranya911 commented Aug 8, 2018

hiranya911 commented Aug 8, 2018

hiranya911 commented Aug 8, 2018

		@@ -0,0 +1,198 @@
		"""SSEClient module to handle streaming of realtime changes on the database

[WIP] Adding stream listener to stream changes in child nodes #183

[WIP] Adding stream listener to stream changes in child nodes #183

Conversation

the-c0d3r commented Jul 21, 2018 • edited

hiranya911 commented Jul 23, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

the-c0d3r commented Jul 26, 2018

hiranya911 commented Jul 26, 2018 • edited

Update:

the-c0d3r commented Jul 27, 2018

hiranya911 commented Jul 27, 2018

hiranya911 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

the-c0d3r Jul 28, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

the-c0d3r commented Jul 28, 2018

hiranya911 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

roboflank commented Jul 31, 2018

hiranya911 commented Aug 2, 2018

hiranya911 left a comment

Choose a reason for hiding this comment

the-c0d3r commented Aug 3, 2018

hiranya911 commented Aug 7, 2018

the-c0d3r commented Aug 8, 2018

hiranya911 commented Aug 8, 2018

hiranya911 commented Aug 8, 2018

hiranya911 commented Aug 8, 2018

the-c0d3r commented Jul 21, 2018 •

edited

hiranya911 commented Jul 23, 2018 •

edited

hiranya911 commented Jul 26, 2018 •

edited

the-c0d3r Jul 28, 2018 •

edited