Add TAsyncCompactProtocol and TAsyncFramedTransport

[aiudirog]

For consistency with the standard implementation, I've added the async versions of the compact protocol and framed transport.

I had to refactor the aio test cases because they were getting unruly. I chose to have a base test case which has all of the generic tests and then a subclass for each combination of protocol, transport, and SSL (the SSL is added as a mixin). If you don't like this design I can find a different way.

---

I also found a very small bug in the compact protocol where booleans with field id 0 weren't being written correctly. As you can see in the code below, the Java implementation checks to see if the bool fid is not null and if so calls their write field header (writeFieldBeginInternal()) where it then checks if it can compact with previous field. The Python code checks if the bool fid is falsey as well as some of the internal conditions for compacting the fields which prevents it from calling _write_field_header() when it should.

Python Source

    def write_bool(self, bool):
        # Use a ternary once instead of separate if statements
        # Changed to make this example easier to read
        ctype = CompactType.TRUE if bool else CompactType.FALSE
        if self._bool_fid and self._bool_fid > self._last_fid \
                and self._bool_fid - self._last_fid <= 15:
            self._write_field_header(ctype, self._bool_fid)
        else:
            self.write_byte(ctype)

    def _write_field_header(self, type, fid):
        delta = fid - self._last_fid
        if 0 < delta <= 15:
            self.write_ubyte(delta << 4 | type)
        else:
            self.write_byte(type)
            self.write_i16(fid)
        self._last_fid = fid

Java Source

  public void writeBool(boolean b) throws TException {
    if (booleanField_ != null) {
      // we haven't written the field header yet
      writeFieldBeginInternal(booleanField_, b ? Types.BOOLEAN_TRUE : Types.BOOLEAN_FALSE);
      booleanField_ = null;
    } else {
      // we're not part of a field, so just write the value.
      writeByteDirect(b ? Types.BOOLEAN_TRUE : Types.BOOLEAN_FALSE);
    }
  }
  private void writeFieldBeginInternal(TField field, byte typeOverride) throws TException {
    // short lastField = lastField_.pop();

    // if there's a type override, use that.
    byte typeToWrite = typeOverride == -1 ? getCompactType(field.type) : typeOverride;

    // check if we can use delta encoding for the field id
    if (field.id > lastFieldId_ && field.id - lastFieldId_ <= 15) {
      // write them together
      writeByteDirect((field.id - lastFieldId_) << 4 | typeToWrite);
    } else {
      // write them separate
      writeByteDirect(typeToWrite);
      writeI16(field.id);
    }

    lastFieldId_ = field.id;
    // lastField_.push(field.id);
  }

If the original intention was to only call _write_field_header() if the compaction was going to happen, this should make it work (but I think the Java implementation is more efficient as it doesn't compute the delta twice):

    def write_bool(self, bool):
        ctype = CompactType.TRUE if bool else CompactType.FALSE
        if self._bool_fid and self._bool_fid > self._last_fid \
                and self._bool_fid - self._last_fid <= 15:
            self._write_field_header(ctype, self._bool_fid)
        else:
            self.write_byte(ctype)
            if self._bool_fid is not None:
                self.write_i16(self._bool_fid)

If you leave off the bug fix commit, you should see that my test cases fail when the protocol needs to return a boolean.

[codecov]

Codecov Report

Merging #103 into master will decrease coverage by 0.73%.
The diff coverage is 66.9%.

@@            Coverage Diff             @@
##           master     #103      +/-   ##
==========================================
- Coverage   80.91%   80.17%   -0.74%     
==========================================
  Files          35       39       +4     
  Lines        3536     3803     +267     
==========================================
+ Hits         2861     3049     +188     
- Misses        675      754      +79

Impacted Files	Coverage Δ
thriftpy2/contrib/aio/server.py	89.47% <0%> (+7.89%)	⬆️
thriftpy2/contrib/aio/protocol/__init__.py	100% <100%> (ø)
thriftpy2/protocol/compact.py	82.22% <100%> (+1.83%)	⬆️
thriftpy2/contrib/aio/transport/__init__.py	100% <100%> (ø)
thriftpy2/contrib/aio/protocol/compact.py	60.81% <60.81%> (ø)
thriftpy2/contrib/aio/transport/framed.py	92.85% <92.85%> (ø)
... and 2 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 3d08210...63850b4. Read the comment docs.

[ethe]

Thank you, this is awesome, I am reviewing it.

[ethe]

Seems like there are some methods do not need to add asyncio.coroutine decorator.

[aiudirog]

I decorated all the methods starting with read to make it easier to convert by assuming all reads were coroutines. I can change it if you like as there is a slight performance cost.

[ethe]

Yes, if the method does not depend on I/O, it is better to remove the coroutine decorator.

[aiudirog]

I removed all the ones I could. TAsyncClient assumes read_message_end() is a coroutine so I left that one with the decorator.

[aiudirog]

It may be worth it to go through all the different protocol classes and mark any helper methods as private so it's easier to determine which need to be async regardless of content based upon the public interface.

[ethe]

Yes, maybe we should promise an interface to describe which methods of a protocol should be coroutines.

[aiudirog]

I can take care of that in a dedicated PR for standardizing interfaces.

[ethe]

I think the refactored test cases are pretty well, thank you.

[ethe]

LGTM except for a small doubt as before

[aiudirog]

While running the tests again, I noticed some warnings that showed the clients and servers weren't being cleaned up properly. I updated the test cases to ensure all clients are closed and I fixed a bug in TAsyncServer.serve() where close() wasn't being awaited.