Оборачивание последовательности в потоке в F #

Question

У меня есть функция, которая принимает поток.Мои данные находятся в большом списке, состоящем из миллионов элементов.

Есть ли простой способ обернуть последовательность в потоке, возвращая куски моей последовательности в потоке?Одним из очевидных подходов является реализация моего собственного потокового класса, который возвращает фрагменты последовательности.Что-то вроде:

type SeqStream(sequence:seq<'a>) = 
    inherit Stream()
    default x.Read(buf, offset, count) =
        // get next chunk
        // yield chunk

Есть ли более простой способ сделать это?У меня нет средств для изменения целевой функции, которая принимает поток, хотя.

Answer 1

Я думаю, что ваш подход выглядит хорошо. Единственная проблема заключается в том, что Stream является относительно сложным классом, в котором довольно много членов, и вы, вероятно, не захотите реализовывать большинство из них - если вы хотите передать его в некоторый код, который использует некоторые дополнительные члены, Вам нужно будет сделать реализацию более сложной. В любом случае, простой поток, который реализует только Read, может выглядеть так:

type SeqStream<'a>(sequence:seq<'a>, formatter:'a -> byte[]) =
  inherit Stream()

  // Keeps bytes that were read previously, but were not used    
  let temp = ResizeArray<_>() 
  // Enumerator for reading data from the sequence
  let en = sequence.GetEnumerator()

  override x.Read(buffer, offset, size) = 
    // Read next element and add it to temp until we have enough
    // data or until we reach the end of the sequence
    while temp.Count < size && en.MoveNext() do
      temp.AddRange(formatter(en.Current))

    // Copy data to the output & return count (may be less then 
    // required (at the end of the sequence)
    let ret = min size temp.Count
    temp.CopyTo(0, buffer, offset, ret)
    temp.RemoveRange(0, ret)
    ret

  override x.Seek(offset, dir) = invalidOp "Seek"
  override x.Flush() = invalidOp "Flush"
  override x.SetLength(l) = invalidOp "SetLength"
  override x.Length = invalidOp "Length"
  override x.Position 
    with get() = invalidOp "Position"
    and set(p) = invalidOp "Position"
  override x.Write(buffer, offset, size) = invalidOp "Write"
  override x.CanWrite = false
  override x.CanSeek = false
  override x.CanRead = true

Обратите внимание, что я добавил дополнительный параметр - функцию для преобразования значения универсального типа в байтовый массив. В общем, трудно преобразовать что-либо в байты (вы можете использовать некоторую сериализацию), так что это, вероятно, проще. Например, для целых чисел вы можете написать:

let stream = new SeqStream<_>([ 1 .. 5 ], System.BitConverter.GetBytes)