_handleIncomingRequestStream(payload) {
		const reqStream = this.pendingReqStreams.get(payload.id);
		let pass = reqStream ? reqStream.stream : undefined;
		let isNew = false;

		if (!payload.stream && !pass && !payload.seq) {
			// It is not a stream data
			return false;
		}

		if (!pass) {
			isNew = true;
			this.logger.debug(
				`<= New stream is received from '${payload.sender}'. Seq: ${payload.seq}`
			);

			// Create a new pass stream
			pass = new Transform({
				// TODO: It's incorrect because the chunks may receive in random order, so it processes an empty meta.
				// Meta is filled correctly only in the 0. chunk.
				objectMode: payload.meta && payload.meta["$streamObjectMode"],
				transform: function (chunk, encoding, done) {
					this.push(chunk);
					return done();
				}
			});

			pass.$prevSeq = -1;
			pass.$pool = new Map();

			this.pendingReqStreams.set(payload.id, { sender: payload.sender, stream: pass });

			pass.on("moleculer-timeout-middleware", timeout => {
				setTimeout(() => {
					this.pendingReqStreams.delete(payload.id);
					this._destroyStreamIfPossible(
						pass,
						`Pending request stream ${payload.id} have been closed by timeout ${timeout} ms`
					);
				}, 1000);
			});
		}

		if (payload.seq > pass.$prevSeq + 1) {
			// Some chunks are late. Store these chunks.
			this.logger.debug(
				`Put the chunk into pool (size: ${pass.$pool.size}). Seq: ${payload.seq}`
			);

			pass.$pool.set(payload.seq, payload);

			// TODO: start timer.
			// TODO: check length of pool.
			// TODO: reset seq

			return null;
		}

		// the next stream chunk received
		pass.$prevSeq = payload.seq;

		if (pass.$prevSeq > 0) {
			if (!payload.stream) {
				// Check stream error
				if (payload.meta && payload.meta["$streamError"]) {
					pass.emit(
						"error",
						this._createErrFromPayload(payload.meta["$streamError"], payload)
					);
				}

				this.logger.debug(
					`<= Stream closing is received from '${payload.sender}'. Seq: ${payload.seq}`
				);

				// End of stream
				pass.end();

				// Remove pending request stream
				this.pendingReqStreams.delete(payload.id);

				return null;
			} else {
				this.logger.debug(
					`<= Stream chunk is received from '${payload.sender}'. Seq: ${payload.seq}`
				);
				pass.write(
					payload.params.type === "Buffer"
						? Buffer.from(payload.params.data)
						: payload.params
				);
			}
		}

		// Check newer chunks in the pool
		if (pass.$pool.size > 0) {
			this.logger.debug(`Has stored packets. Size: ${pass.$pool.size}`);
			const nextSeq = pass.$prevSeq + 1;
			const nextPacket = pass.$pool.get(nextSeq);
			if (nextPacket) {
				pass.$pool.delete(nextSeq);
				setImmediate(() => this.requestHandler(nextPacket));
			}
		}

		return pass && payload.seq == 0 ? pass : null;
	}
Handle incoming request stream.

@param {Object} payload
@returns {Stream}