-
Notifications
You must be signed in to change notification settings - Fork 8
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Streams3 adaptor(s) #35
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good work!!!! I've left some notes.
helpers/bob-duplex.js
Outdated
this.cork() | ||
} | ||
|
||
_write(chunk, encoding, callback) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we should add support for _writev
as well.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think that should probably wait for if bob supports multi buffer next()
s or something. Otherwise it seems more reasonable to just have WritableStream
manage it?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
+1. We should add multi buffed support.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should be discussed at #30
helpers/bob-duplex.js
Outdated
if (this.source !== null) { | ||
this.source.pull(err) | ||
this[kDestroyCallback] = cb | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Need an else here. If there is no source BUT sink exists then we are responsible for calling this.sink.next(error, status.error, ...)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Actually you have to destroy both the sink and the source always and wait for them both to be finished destroying ...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think it should just explode if a source doesn't exist, tbh.
tests/bob-streams3-from-source.js
Outdated
const fileSource = new FileSource(process.argv[2]) | ||
const bobDuplex = new BobDuplex({ highWaterMark: 1024, name: '1' }) | ||
|
||
new Stream(fileSource, bobDuplex) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This BobDuplex
API is confusing to me. I think a more common API would be new BobDuplex({ hwm, name, sink, source })
.
Calling bindSource()
should error because it should be in the constructor instead. Calling bindSink()
is fine because that's part of the implementors API and it's necessary to be a valid bob.Source
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah agree it's not ideal
helpers/bob-duplex.js
Outdated
this.sink.next(status_type.continue, null, chunk, chunk.length) | ||
} | ||
|
||
_read (size) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm worried _read()
will not get called again if we do not call push()
with enough data ...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@Raynos I don't really understand... Could you elaborate?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The documentation of Readable
Seems to imply that if you do not push at least size bytes it will not call _read again.
The current implementation seems to be doing one pull call and it might return <size bytes
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@mcollina could you possibly confirm this?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The docs seem to say:
The
size
argument is advisory.
helpers/bob-duplex.js
Outdated
|
||
// Regular data case. | ||
if (status === status_type.continue) { | ||
this.push(buffer.slice(0, bytes)) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
the return type of push()
is indicating whether we need to call pull()
again and it's currently ignored. I'm worried this stream will get stuck in a weird state where Readable
will not call _read
again because you didn't push enough data.
The behavior of Readable
is different in the use case where pipe()
is called vs read()
+ readable.on('readable', ...)
etc
helpers/bob-duplex.js
Outdated
this.sink = sink | ||
} | ||
|
||
next (status, error, buffer, bytes) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm pretty sure that if bytes < buffer.length
we are supposed to call pull()
again for this to be a correctly implemented Readable
with _read
helpers/bob-duplex.js
Outdated
} | ||
|
||
// If we have an error and have not already been in error state, emit it Streams3 style. | ||
if (error !== null) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
checking status === status_type.error
is nicer then check if error !== null
I started reading this, readable-stream, fs-sink, fs-source & pull-stream-to-stream I read some comments in readable-stream that I knew were there 5 years ago and I read them in isaacs voice, there are many mode and many tests to write. Please use https://github.com/dominictarr/stream-spec and friends. These data back to 2015 and was in the streams npm era where readable-stream did not exist and we were out of luck, we had to implement Readable, Writable & Duplex by hand. We had to implement the really old streams1 api of 'data', 'end', write, end, pipe, etc. There is no |
Talking of spec, can we port these to bob I believe Verify is a partial implementation of it. |
@Raynos What if this extended / was the (bob) It looks like e,g, |
7f27eca
to
f48c3f6
Compare
Updated with added automated tests! |
551ea5e
to
8ec6b97
Compare
f48c3f6
to
aa251c1
Compare
Ok so since there's some comments here about the API design in general, here's The Dream:
Basically this isn't ideal but it is an important between step. |
aa251c1
to
0fe264f
Compare
9d2a491
to
c5a7502
Compare
We will need BobDuplex for backcompat. I imagine for node core we want to expose new APIs that use bob streams and deprecate streams3 APIs. I don't think we can make a breaking change, we will always need streams3 and thus Also for things like |
I think you're maybe missing the idea. The idea is and always has been to implement Which is probably the only way we can actually get a reasonable transition to happen. |
|
Correct, but it is technically then the wrong way around - it takes a Using that then, you can stick it "inbetween" a bob stream and a streams3 stream to connect them. Ideally though, that wouldn't be necessary - you'd just set your stream up and it both would be a streams3 stream and also streams3 streams would have the bob api "underneath". |
0fe264f
to
50f2c80
Compare
50f2c80
to
7c86655
Compare
7cbf4e1
to
985f2f5
Compare
Tests pass, seems unnecessary as per comments. Just in case, though, I want to keep these in-place.
yeah, there's no standard way to do this
this is much better! \o/
for version pinning stability
Should handle things a bit better.
This error handling seems more correct to me.
Ok, clean up pushed. Error handling and some other state stuff should be more correct now, I think. |
const writeCb = this[kWriteCallback] | ||
this[kWriteCallback] = null | ||
|
||
// If we don't yet have a write callback from an attempted Streams3 write, just bail. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
One thing I don't know what to do about though, is this.
I can't get it to work without bailing in this case no matter how the streams are composed in user code...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Readable class looks legit, a bunch of comments for the Writable class.
this[kPulling] = false | ||
|
||
// Send data to our sink. | ||
this.sink.next(status_type.continue, null, chunk, chunk.length) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we want to implement buffering here ? Same in the read case. We may already be "piping" data before we have hooked up the sinks and sources.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If not; add a check for this.sink
and throw some kind of readable error about writing to stream before new Stream()
or source.bindSink()
was called.
const encoding = this._readableState.encoding || this._readableState.defaultEncoding | ||
chunk = Buffer.from(chunk, encoding) | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I suspect this needs an if check here if (this[kPulling]) { cb() } else { this[kWriteCallbac] = callback }
this[kPulling] = false | ||
|
||
// Send data to our sink. | ||
this.sink.next(status_type.continue, null, chunk, chunk.length) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If this[kPulling]
is false
here then we do not want to call this.sink.next()
yet; we want to store the current chunk
in this[kPendingChunk]
Landed |
It works!!
ReadableSink
to turn BOB flow to streams3,WritableSource
to turn streams3 flow into BOB.Implements #18
Test cases: run
npm test
!(manually verify the output file,readme_
orreadme.gz
(there is an ungzip helper intests/
).(may require Node 13)
Use
NODE_DEBUG=bob
to see what is happening.Replacereadme.md
with your choice of file.R=@mcollina