如何拆分和传输多个 NAudio 流

Question

我有一个 C# 项目使用来自 Kinect 1、Kinect 2、麦克风或其他任何设备的输入音频流。

waveIn.DataAvailable += (object sender, WaveInEventArgs e) => {
  lock(buffer){
    var pos = buffer.Position;
              buffer.Write(e.Buffer, 0, e.BytesRecorded);
              buffer.Position = pos;
  }
};

缓冲区变量是来自组件 A 的 Stream，它将由处理 Streams 的 SpeechRecognition 组件 B 处理。

我将添加新组件 C、D、E，在 Streams 上工作以计算音调、检测声音、进行指纹识别或其他任何事情...

如何为组件 C、D、E 复制流？

组件A发送事件"I have a Stream do what you want"我不想通过事件反转逻辑"Give me your streams"
我正在寻找一个 "MultiStream" 可以给我一个 Stream 实例并且可以处理工作的

组件 A

var MultiStream buffer = new MultiStream()
...
SendMyEventWith(buffer)

组件 B、C、D、E

public void HandleMyEvent(MultiStream buffer){
  var stream = buffer.GetNewStream();
  var engine = new EngineComponentB()
      engine.SetStream(stream);
}

MultiStream 必须是一个 Stream 来包装 Write() 方法（因为 Stream 没有数据可用机制）？
如果流是组件 B 的 Dispose()，MultiStream 应该将其从数组中删除吗？
MultiStream 必须在 Read() 上抛出异常才能要求使用 GetNewStream()

编辑：Kinect 1 本身提供一个 Stream ... :-( 我应该使用 Thread 将其泵入 MultiStream 吗？

有人有那种 MultiStream Class 吗？

谢谢

Answer 1

不知何故，我认为流并不适合您想要做的事情。您正在设置一个很长的运行程序将无缘无故地不断扩展数据需求的情况。

我建议使用 pub/sub 模型将接收到的音频数据发布给订阅者，最好使用多线程方法来最大程度地减少不良订阅者的影响。可以找到一些想法 here.

我之前用处理器 class 实现了 IObserver<byte[]> 并使用 Queue<byte[]> 存储样本块，直到进程线程为它们准备好。这是基础 classes:

public abstract class BufferedObserver<T> : IObserver<T>, IDisposable
{
    private object _lck = new object();

    private IDisposable _subscription = null;
    public bool Subscribed { get { return _subscription != null; } }

    private bool _completed = false;
    public bool Completed { get { return _completed; } }

    protected readonly Queue<T> _queue = new Queue<T>();

    protected bool DataAvailable { get { lock(_lck) { return _queue.Any(); } } }
    protected int AvailableCount { get { lock (_lck) { return _queue.Count; } } }

    protected BufferedObserver()
    {
    }

    protected BufferedObserver(IObservable<T> observable)
    {
        SubscribeTo(observable);
    }

    public virtual void Dispose()
    {
        if (_subscription != null)
        {
            _subscription.Dispose();
            _subscription = null;
        }
    }

    public void SubscribeTo(IObservable<T> observable)
    {
        if (_subscription != null)
            _subscription.Dispose();
        _subscription = observable.Subscribe(this);
        _completed = false;
    }

    public virtual void OnCompleted()
    {
        _completed = true;
    }

    public virtual void OnError(Exception error)
    { }

    public virtual void OnNext(T value)
    {
        lock (_lck)
            _queue.Enqueue(value);
    }

    protected bool GetNext(ref T buffer)
    {
        lock (_lck)
        {
            if (!_queue.Any())
                return false;
            buffer = _queue.Dequeue();
            return true;
        }
    }

    protected T NextOrDefault()
    {
        T buffer = default(T);
        GetNext(ref buffer);
        return buffer;
    }
}

public abstract class Processor<T> : BufferedObserver<T>
{
    private object _lck = new object();
    private Thread _thread = null;

    private object _cancel_lck = new object();
    private bool _cancel_requested = false;
    private bool CancelRequested
    {
        get { lock(_cancel_lck) return _cancel_requested; }
        set { lock(_cancel_lck) _cancel_requested = value; }
    }

    public bool Running { get { return _thread == null ? false : _thread.IsAlive; } }
    public bool Finished { get { return _thread == null ? false : !_thread.IsAlive; } }

    protected Processor(IObservable<T> observable)
        : base(observable)
    { }

    public override void Dispose()
    {
        if (_thread != null && _thread.IsAlive)
        {
            //CancelRequested = true;
            _thread.Join(5000);
        }
        base.Dispose();
    }

    public bool Start()
    {
        if (_thread != null)
            return false;

        _thread = new Thread(threadfunc);
        _thread.Start();
        return true;
    }

    private void threadfunc()
    {
        while (!CancelRequested && (!Completed || _queue.Any()))
        {
            if (DataAvailable)
            {
                T data = NextOrDefault();
                if (data != null && !data.Equals(default(T)))
                    ProcessData(data);
            }
            else
                Thread.Sleep(10);
        }
    }

    // implement this in a sub-class to process the blocks
    protected abstract void ProcessData(T data);
}

通过这种方式，您只在需要时才保留数据，并且您可以根据需要将任意数量的进程线程附加到同一个可观察数据源。

为了完整起见，这里有一个实现 IObservable<T> 的通用 class，因此您可以看到它们是如何组合在一起的。这个竟然还有评论：

/// <summary>Generic IObservable implementation</summary>
/// <typeparam name="T">Type of messages being observed</typeparam>
public class Observable<T> : IObservable<T>
{
    /// <summary>Subscription class to manage unsubscription of observers.</summary>
    private class Subscription : IDisposable
    {
        /// <summary>Observer list that this subscription relates to</summary>
        public readonly ConcurrentBag<IObserver<T>> _observers;

        /// <summary>Observer to manage</summary>
        public readonly IObserver<T> _observer;

        /// <summary>Initialize subscription</summary>
        /// <param name="observers">List of subscribed observers to unsubscribe from</param>
        /// <param name="observer">Observer to manage</param>
        public Subscription(ConcurrentBag<IObserver<T>> observers, IObserver<T> observer)
        {
            _observers = observers;
            _observer = observer;
        }

        /// <summary>On disposal remove the subscriber from the subscription list</summary>
        public void Dispose()
        {
            IObserver<T> observer;
            if (_observers != null && _observers.Contains(_observer))
                _observers.TryTake(out observer);
        }
    }

    // list of subscribed observers
    private readonly ConcurrentBag<IObserver<T>> _observers = new ConcurrentBag<IObserver<T>>();

    /// <summary>Subscribe an observer to this observable</summary>
    /// <param name="observer">Observer instance to subscribe</param>
    /// <returns>A subscription object that unsubscribes on destruction</returns>
    /// <remarks>Always returns a subscription.  Ensure that previous subscriptions are disposed
    /// before re-subscribing.</remarks>
    public IDisposable Subscribe(IObserver<T> observer)
    {
        // only add observer if it doesn't already exist:
        if (!_observers.Contains(observer))
            _observers.Add(observer);

        // ...but always return a new subscription.
        return new Subscription(_observers, observer);
    }

    // delegate type for threaded invocation of IObserver.OnNext method
    private delegate void delNext(T value);

    /// <summary>Send <paramref name="data"/> to the OnNext methods of each subscriber</summary>
    /// <param name="data">Data object to send to subscribers</param>
    /// <remarks>Uses delegate.BeginInvoke to send out notifications asynchronously.</remarks>
    public void Notify(T data)
    {
        foreach (var observer in _observers)
        {
            delNext handler = observer.OnNext;
            handler.BeginInvoke(data, null, null);
        }
    }

    // delegate type for asynchronous invocation of IObserver.OnComplete method
    private delegate void delComplete();

    /// <summary>Notify all subscribers that the observable has completed</summary>
    /// <remarks>Uses delegate.BeginInvoke to send out notifications asynchronously.</remarks>
    public void NotifyComplete()
    {
        foreach (var observer in _observers)
        {
            delComplete handler = observer.OnCompleted;
            handler.BeginInvoke(null, null);
        }
    }
}

现在您可以创建一个 Observable<byte[]> 作为您感兴趣的 Process<byte[]> 个实例的传送器。从输入流、音频reader等中拉出数据块，传递给Notify方法。只需确保事先克隆阵列...

Answer 2

我不确定这是最好的方法还是比以前的答案更好，而且我不认为运行这段代码是完美的，但我编码了一些东西这正是您所要求的，因为它很有趣 - MultiStream class.

您可以在此处找到 class 的代码：http://pastie.org/10289142

用法示例：

MultiStream ms = new MultiStream();

Stream copy1 = ms.CloneStream();
ms.Read( ... );

Stream copy2 = ms.CloneStream();
ms.Read( ... );

copy1 和 copy2 将在示例运行之后包含相同的数据，并且它们将随着 MultiStream 的写入而继续更新。您可以单独读取、更新位置和处理克隆的流。如果处置，克隆的流将从 MultiStream 中删除，处置 Multistream 将关闭所有相关和克隆的流（如果这不是您想要的行为，您可以更改它）。尝试写入克隆的流将引发不受支持的异常。

如何拆分和传输多个 NAudio 流

How can I split and pipe multiple NAudio stream

c#

stream

naudio