• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1'use strict';
2const common = require('../common');
3const assert = require('assert');
4const Readable = require('stream').Readable;
5
6{
7  const readable = new Readable({
8    read(size) {}
9  });
10
11  const state = readable._readableState;
12
13  // Starting off with false initially.
14  assert.strictEqual(state.reading, false);
15  assert.strictEqual(state.readingMore, false);
16
17  readable.on('data', common.mustCall((data) => {
18    // While in a flowing state with a 'readable' listener
19    // we should not be reading more
20    if (readable.readableFlowing)
21      assert.strictEqual(state.readingMore, true);
22
23    // Reading as long as we've not ended
24    assert.strictEqual(state.reading, !state.ended);
25  }, 2));
26
27  function onStreamEnd() {
28    // End of stream; state.reading is false
29    // And so should be readingMore.
30    assert.strictEqual(state.readingMore, false);
31    assert.strictEqual(state.reading, false);
32  }
33
34  const expectedReadingMore = [true, true, false];
35  readable.on('readable', common.mustCall(() => {
36    // There is only one readingMore scheduled from on('data'),
37    // after which everything is governed by the .read() call
38    assert.strictEqual(state.readingMore, expectedReadingMore.shift());
39
40    // If the stream has ended, we shouldn't be reading
41    assert.strictEqual(state.ended, !state.reading);
42
43    // Consume all the data
44    while (readable.read() !== null);
45
46    if (expectedReadingMore.length === 0) // Reached end of stream
47      process.nextTick(common.mustCall(onStreamEnd, 1));
48  }, 3));
49
50  readable.on('end', common.mustCall(onStreamEnd));
51  readable.push('pushed');
52
53  readable.read(6);
54
55  // reading
56  assert.strictEqual(state.reading, true);
57  assert.strictEqual(state.readingMore, true);
58
59  // add chunk to front
60  readable.unshift('unshifted');
61
62  // end
63  readable.push(null);
64}
65
66{
67  const readable = new Readable({
68    read(size) {}
69  });
70
71  const state = readable._readableState;
72
73  // Starting off with false initially.
74  assert.strictEqual(state.reading, false);
75  assert.strictEqual(state.readingMore, false);
76
77  readable.on('data', common.mustCall((data) => {
78    // While in a flowing state without a 'readable' listener
79    // we should be reading more
80    if (readable.readableFlowing)
81      assert.strictEqual(state.readingMore, true);
82
83    // Reading as long as we've not ended
84    assert.strictEqual(state.reading, !state.ended);
85  }, 2));
86
87  function onStreamEnd() {
88    // End of stream; state.reading is false
89    // And so should be readingMore.
90    assert.strictEqual(state.readingMore, false);
91    assert.strictEqual(state.reading, false);
92  }
93
94  readable.on('end', common.mustCall(onStreamEnd));
95  readable.push('pushed');
96
97  // Stop emitting 'data' events
98  assert.strictEqual(state.flowing, true);
99  readable.pause();
100
101  // paused
102  assert.strictEqual(state.reading, false);
103  assert.strictEqual(state.flowing, false);
104
105  readable.resume();
106  assert.strictEqual(state.reading, false);
107  assert.strictEqual(state.flowing, true);
108
109  // add chunk to front
110  readable.unshift('unshifted');
111
112  // end
113  readable.push(null);
114}
115
116{
117  const readable = new Readable({
118    read(size) {}
119  });
120
121  const state = readable._readableState;
122
123  // Starting off with false initially.
124  assert.strictEqual(state.reading, false);
125  assert.strictEqual(state.readingMore, false);
126
127  const onReadable = common.mustNotCall();
128
129  readable.on('readable', onReadable);
130
131  readable.on('data', common.mustCall((data) => {
132    // Reading as long as we've not ended
133    assert.strictEqual(state.reading, !state.ended);
134  }, 2));
135
136  readable.removeListener('readable', onReadable);
137
138  function onStreamEnd() {
139    // End of stream; state.reading is false
140    // And so should be readingMore.
141    assert.strictEqual(state.readingMore, false);
142    assert.strictEqual(state.reading, false);
143  }
144
145  readable.on('end', common.mustCall(onStreamEnd));
146  readable.push('pushed');
147
148  // We are still not flowing, we will be resuming in the next tick
149  assert.strictEqual(state.flowing, false);
150
151  // Wait for nextTick, so the readableListener flag resets
152  process.nextTick(function() {
153    readable.resume();
154
155    // Stop emitting 'data' events
156    assert.strictEqual(state.flowing, true);
157    readable.pause();
158
159    // paused
160    assert.strictEqual(state.flowing, false);
161
162    readable.resume();
163    assert.strictEqual(state.flowing, true);
164
165    // add chunk to front
166    readable.unshift('unshifted');
167
168    // end
169    readable.push(null);
170  });
171}
172