function initializeReadableStream(underlyingSource, strategy)
{
"use strict";
if (underlyingSource === @undefined)
underlyingSource = { };
if (strategy === @undefined)
strategy = { highWaterMark: 1, size: function() { return 1; } };
if (!@isObject(underlyingSource))
@throwTypeError("ReadableStream constructor takes an object as first argument");
if (strategy !== @undefined && !@isObject(strategy))
@throwTypeError("ReadableStream constructor takes an object as second argument, if any");
this.@state = @streamReadable;
this.@reader = @undefined;
this.@storedError = @undefined;
this.@disturbed = false;
this.@readableStreamController = null;
const type = underlyingSource.type;
const typeString = @String(type);
if (typeString === "bytes") {
if (strategy.highWaterMark === @undefined)
strategy.highWaterMark = 0;
try {
let readableByteStreamControllerConstructor = @ReadableByteStreamController;
} catch (e) {
@throwTypeError("ReadableByteStreamController is not implemented");
}
this.@readableStreamController = new @ReadableByteStreamController(this, underlyingSource, strategy.highWaterMark);
} else if (type === @undefined) {
if (strategy.highWaterMark === @undefined)
strategy.highWaterMark = 1;
this.@readableStreamController = new @ReadableStreamDefaultController(this, underlyingSource, strategy.size, strategy.highWaterMark);
} else
@throwRangeError("Invalid type for underlying source");
return this;
}
function cancel(reason)
{
"use strict";
if (!@isReadableStream(this))
return @Promise.@reject(@makeThisTypeError("ReadableStream", "cancel"));
if (@isReadableStreamLocked(this))
return @Promise.@reject(new @TypeError("ReadableStream is locked"));
return @readableStreamCancel(this, reason);
}
function getReader(options)
{
"use strict";
if (!@isReadableStream(this))
throw @makeThisTypeError("ReadableStream", "getReader");
if (options === @undefined)
options = { };
if (options.mode === 'byob') {
@throwTypeError("ReadableStreamBYOBReader is not implemented");
}
if (options.mode === @undefined)
return new @ReadableStreamDefaultReader(this);
@throwRangeError("Invalid mode is specified");
}
function pipeThrough(streams, options)
{
"use strict";
const writable = streams.writable;
const readable = streams.readable;
this.pipeTo(writable, options);
return readable;
}
function pipeTo(destination)
{
"use strict";
const options = arguments[1];
const preventClose = @isObject(options) && !!options.preventClose;
const preventAbort = @isObject(options) && !!options.preventAbort;
const preventCancel = @isObject(options) && !!options.preventCancel;
const source = this;
let reader;
let lastRead;
let lastWrite;
let closedPurposefully = false;
let promiseCapability;
function doPipe() {
lastRead = reader.read();
@Promise.prototype.@then.@call(@Promise.all([lastRead, destination.ready]), function([{ value, done }]) {
if (done)
closeDestination();
else if (destination.state === "writable") {
lastWrite = destination.write(value);
doPipe();
}
}, function(e) {
throw e;
});
}
function cancelSource(reason) {
if (!preventCancel) {
reader.cancel(reason);
reader.releaseLock();
promiseCapability.@reject.@call(@undefined, reason);
} else {
@Promise.prototype.@then.@call(lastRead, function() {
reader.releaseLock();
promiseCapability.@reject.@call(@undefined, reason);
});
}
}
function closeDestination() {
reader.releaseLock();
const destinationState = destination.state;
if (!preventClose && (destinationState === "waiting" || destinationState === "writable")) {
closedPurposefully = true;
@Promise.prototype.@then.@call(destination.close(), promiseCapability.@resolve, promiseCapability.@reject);
} else if (lastWrite !== @undefined)
@Promise.prototype.@then.@call(lastWrite, promiseCapability.@resolve, promiseCapability.@reject);
else
promiseCapability.@resolve.@call();
}
function abortDestination(reason) {
reader.releaseLock();
if (!preventAbort)
destination.abort(reason);
promiseCapability.@reject.@call(@undefined, reason);
}
promiseCapability = @newPromiseCapability(@Promise);
reader = source.getReader();
@Promise.prototype.@then.@call(reader.closed, @undefined, abortDestination);
@Promise.prototype.@then.@call(destination.closed,
function() {
if (!closedPurposefully)
cancelSource(new @TypeError('destination is closing or closed and cannot be piped to anymore'));
},
cancelSource
);
doPipe();
return promiseCapability.@promise;
}
function tee()
{
"use strict";
if (!@isReadableStream(this))
throw @makeThisTypeError("ReadableStream", "tee");
return @readableStreamTee(this, false);
}
function locked()
{
"use strict";
if (!@isReadableStream(this))
throw @makeGetterTypeError("ReadableStream", "locked");
return @isReadableStreamLocked(this);
}