Skip to content

Commit 36bd4ea

Browse files
fix(interceptTransfer): rename reworkers to operators
1 parent 5ca78e4 commit 36bd4ea

File tree

1 file changed

+13
-13
lines changed

1 file changed

+13
-13
lines changed

packages/operators/src/transfer/interceptTransfer.js

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -2,14 +2,14 @@ import { concatMap, from, map, of } from 'rxjs';
22

33
import { readBytes } from './utils';
44

5-
export const interceptTransfer = (reworkers = [], chunkSize = 60 * 1024) => {
5+
export const interceptTransfer = (operators = [], chunkSize = 60 * 1024) => {
66
return source =>
77
source.pipe(
88
concatMap(requestResponse => {
9-
if (reworkers.length) {
9+
if (operators.length) {
1010
return of(requestResponse).pipe(
1111
sourceToStream(),
12-
interceptStream(reworkers, chunkSize),
12+
interceptStream(operators, chunkSize),
1313
streamToSource(requestResponse)
1414
);
1515
}
@@ -28,7 +28,7 @@ const streamToSource = reqResp => {
2828
source.pipe(concatMap(stream => streamToObjectMap.get(reqResp.constructor)(stream, reqResp)));
2929
};
3030

31-
const interceptStream = (reworkers, chunkSize) => {
31+
const interceptStream = (operators, chunkSize) => {
3232
return source =>
3333
source.pipe(
3434
map(({ stream, total }) => {
@@ -44,9 +44,9 @@ const interceptStream = (reworkers, chunkSize) => {
4444
const { done, value } = await this.chunks.next();
4545
try {
4646
if (done) {
47-
return onStreamEnd(controller, reworkers);
47+
return onStreamEnd(controller, operators);
4848
}
49-
await onStreamPull(controller, reworkers, value, total, this.time);
49+
await onStreamPull(controller, operators, value, total, this.time);
5050
} catch (err) {
5151
onStreamError(err);
5252
throw err;
@@ -103,17 +103,17 @@ const convertStreamToResponse = (stream, resp) => {
103103
);
104104
};
105105

106-
const onStreamPull = async (controller, reworkers, value, total, period) => {
107-
controller.enqueue(new Uint8Array(value));
108-
reworkers.map(reworker => reworker.next({ value, total, period }));
106+
const onStreamPull = async (controller, operators, bytes, total, period) => {
107+
controller.enqueue(new Uint8Array(bytes));
108+
operators.map(operator => operator.next({ value: bytes, total, period }));
109109
};
110110

111-
const onStreamEnd = (controller, reworkers) => {
111+
const onStreamEnd = (controller, operators) => {
112112
controller.close();
113-
reworkers.map(reworker => reworker.complete());
113+
operators.map(operator => operator.complete());
114114
return;
115115
};
116116

117-
const onStreamError = (reworkers, err) => {
118-
reworkers.map(reworker => reworker.error(err));
117+
const onStreamError = (operators, err) => {
118+
operators.map(operator => operator.error(err));
119119
};

0 commit comments

Comments
 (0)