本文将介绍如何利用IEnumerable接口创建StreamInsight应用程序。

IEnumerable与IObservable最大的区别在于IEnumerable是采用Pull模式获取数据,而IObservable则是用Push模式推送数据。

用户首先通过IEnumerable.GetEnumerator来获得数据源的Enumerator,然后通过调用IEnumerator.MoveNext和IEnumerator.Current来获取数据。StreamInsight的输入输出通过一个数据队列连接。输入适配器获得数据,并调用Enqueue将数据放入队列,输出适配器循环检测队列是否为空,如果不是空队列,则调用Dequeue将数据取出。

在HelloInsight程序基础上,可以通过如下修改来实现IEnumerable接口。

1、获取输入数据源的Enumerator(HelloPointInput.cs):

public HelloPointInput(HelloInputConfig config)

{

    _config = config;

 

    var streamReader = new StreamReader(config.fileName);

    strings = new List<string>();

           

    while (!streamReader.EndOfStream)

    {

        strings.Add(streamReader.ReadLine());

    }

    stringEnumerator = strings.GetEnumerator();

           

    streamReader.Close();

}

2、读取Enumerator中的数据,并放入队列。注意判断队满的情况(HelloPointInput.cs):

private void ProduceEvents()

{

    while (AdapterState != AdapterState.Stopping)

    {

        if (stringEnumerator.MoveNext())

        {

            try

            {

                var line = stringEnumerator.Current;

                var pendingEvent = CreateInsertEvent();

                pendingEvent.StartTime = DateTime.Now;

                pendingEvent.Payload = new HelloPayload

                {

                    str = line

                };

                EnqueueOperationResult result;

 

                result = Enqueue(ref pendingEvent);

                if (result == EnqueueOperationResult.Full)

                {

                    Thread.Sleep(1000);

                    //Ready();

                    return;

                }

                EnqueueCtiEvent(DateTime.Now);

                Thread.Sleep(1000);

            }

            catch

            {

                //error handling should go here

            }

        }

        else

        {

            break;

        }

    }

    Stopped();

}

注意:在IEnumerator模式下,CTI事件是必须的,否则不会产生数据输出。

3、循环读取队列中的数据,并进行输出处理(HelloPointOutput.cs):

private void ConsumeEvents()

{

    PointEvent<HelloPayload> currEvent;

    DequeueOperationResult result;

 

    while (AdapterState != AdapterState.Stopping)

    {

        result = Dequeue(out currEvent);

 

        if (result == DequeueOperationResult.Empty)

        {

            Ready();

            return;

                                                        

        }

        else

        {

            if (currEvent.EventKind == EventKind.Insert)

            {

                Console.WriteLine("Output: " +

                currEvent.Payload.str

                );

 

            }

            ReleaseEvent(ref currEvent);

        }

    }

 

    Stopped();

}

4、运行程序

建立测试文件:

将输入数据源指向测试文件的存储地址。运行程序,得到如下输出:

 

软件测试工程师 金晶