Async decoding

[sergeyzenchenko]

Hi @gfx I've been looking at async decoding code and looks like it's not doing stream decoding, but instead just tries to decode buffer until it's fully available. I am right?

I think we need to add actual stream decoding. It can speed up decoding significantly for large payloads.

[sergeyzenchenko]

I think I got how async part works, but still it will only deliver object when it's fully available. Have you thought about solution that will deliver parts of the response as soon as they are available? For example if we are parsing array it can send back array elements as soon as they are available.

[sergeyzenchenko]

Something like this:

readArraySize(): number {
    const headByte = this.readHeadByte();

    switch (headByte) {
      case 0xdc:
        return this.readU16();
      case 0xdd:
        return this.readU32();
      default:
        return headByte - 0x90;
    }
  }

  async decodeArrayStream(stream: AsyncIterable<ArrayLike<number> | Uint8Array>, callback: any) {
    let headerParsed = false;
    let decoded = false;
    let itemsLeft = 0;

    for await (const buffer of stream) {
      if (decoded) {
        throw this.createNoExtraBytesError(this.totalPos);
      }

      if (this.headByte === HEAD_BYTE_REQUIRED && !this.hasRemaining()) {
        this.setBuffer(buffer);
      } else {
        // retried because data is insufficient
        const remainingData = this.bytes.subarray(this.pos);
        const newData = ensureUint8Array(buffer);
        const concated = new Uint8Array(remainingData.length + newData.length);
        concated.set(remainingData);
        concated.set(newData, remainingData.length);
        this.setBuffer(concated);
      }

      if (!headerParsed) {
        itemsLeft = this.readArraySize();
        headerParsed = true;
        this.complete();
      }

      try {
        while (true) {
          let result = this.decodeSync();

          callback(result);

          itemsLeft--;

          if (itemsLeft === 0) {
            decoded = true;
            break;
          }
        }
      } catch (e) {
        if (!(e instanceof DataViewIndexOutOfBoundsError)) {
          throw e; // rethrow
        }
        // fallthrough
      }
      this.totalPos += this.pos;
    }
  }

[sergeyzenchenko]

so what do you think about this way? @gfx

[gfx]

Hmm, do you want to use a partial, imcomplete object? I can't imagine its use-cases.

I think partial objects are unstable and unpredictable.

Instead, I have a plan to add something like decodeMultiAsync(), which emits a list of independent MessagePack items one by one.

For example, 0xc3 is a complete MessagePack item that emits true, here isUint8Array.from([0xc3, 0xc3, 0xc3]), so the code like this:

// just a concept code:
const multiItems = Uint8Array.from([0xc3, 0xc3, 0xc3]);
const decoder = decodeMultiAsync(multiItems);
for await (const item of decoder) {
  console.log(item);
}

shows:

true
true
true

What do you think of it?

[gfx]

I think streaming decoding is a killer-feature of MessagePack (and @msgpack/msgpack), but I'll use a complete decoded object for each encoded binary item.

[sergeyzenchenko]

I am thinking about decoding specifically for arrays. So it will emit independent items from array.
For example we are loading array with 10mb of message pack payloads.
And I want to start processing items as soon as they arrive.

[sergeyzenchenko]

Check my PR #42

[sergeyzenchenko]

It does exactly that you said.

[gfx]

Resolved.