Changes to initial drain implementation

[aricart]

connection#drain() to close the connection when drain resolves. connection#drain() also closes the connection when completed