[C++] Make public version of StreamDecoderImpl-like class

[MrLolthe1st]

Describe the enhancement requested

It is useful to use StreamDecoderImpl outside reader.cc.
For example, someone produces stream of data like that:

<SCHEMA>
<RECORD BATCH 0>
...
<RECORD BATCH n - 1>
<EOS> (*)
<SCHEMA>
<DICTIONARY 0>
...
<DICTIONARY k - 1>
<RECORD BATCH 0>
...
<RECORD BATCH n - 1>
<EOS>
<SCHEMA>
<RECORD BATCH 0>
...
<RECORD BATCH n - 1>
<EOS>

StreamDecoder in that way will ignore all payload after (*). But in way where StreamDecoderImpl-like class is open to use, I can use OnMessageDecoded(..) method and delegate message decoding in outer scope and when EOS coming, reseting StreamDecoderImpl to new.

Component(s)

C++

[kou]

How about adding arrow::ipc::StreamDecoder::Reset() instead of exporting StreamDecoderImpl?

[MrLolthe1st]

How about adding arrow::ipc::StreamDecoder::Reset() instead of exporting StreamDecoderImpl?

It's just one approach. In addition to arrow::ipc::StreamDecoder::Reset() must be added arrow::ipc::StreamDecoder::OnMessageDecoded(...), because arrow::ipc::StreamDecoder::Consume(...) doesn't returns amount of readed bytes, so anyway hight-level control of messages must be provided to feed StreamDecoder messages one-by-one.
For example, I'm trying to feed StreamDecoder with payload looks like that:

<RECORD BATCH n - 1>
<EOS>
<SCHEMA>
<DICTIONARY 0>

In that way I'm must determine where message with schema is located (to feed resetted StreamDecoder with new messages, starting with schema message).

Another approach is to make Consume() return not just status, but amount of bytes readed before EOS.

[kou]

How about calling arrow::ipc::StreamDecoder::Reset() in a listener?

class EndlessListener : public arrow::ipc::Listener {
 public:
  explicit EndlessListener() : arrow::ipc::Listener(), decoders_() {}

  arrow::Status OnEOS() override {
    for (auto decoder : decoders_) {
      ARROW_RETURN_NOT_OK(decoder->Reset());
    }
    return arrow::Status::OK();
  }

  void AddDecoder(arrow::ipc::StreamDecoder *decoder) {
    decoders_.push_back(decoder);
  }

 private:
  std::vector<arrow::ipc::StreamDecoder> decoders_;
};

auto listener = std::make_shared<EndlessListener>();
arrow::ipc::StreamDecoder decoder (listener);
listener->AddDecoder(*decoder);
decoder.Consume(...);

[MrLolthe1st]

That's awesome!

[MrLolthe1st]

(moved to #37970 (comment))