问题

我有一个具有时间戳的可观察项（IObservable）。
我使用Scan方法包装每个项并添加对接收到的最后一个有效包装项的引用。
```csharp
IObservable<IWrappedItem> wrappedItems = 
  allItems.Scan(seedItem, 
    (lastWrappedItem, currentItem) => 
      WrapItem(currentItem, lastWrappedItem)));

这是WrapItem的签名：

IWrappedItem WrapItem(Item currentItem, IWrappedItem lastItem);

我们需要更改WrapItem方法，以便跳过无效（非空）的项并返回null。
seedItem很可能为null，而WrapItem方法可以处理它。

我需要更新对Scan的使用方式，类似于这样：

IObservable<IWrappedItem> wrappedItems = allItems.Scan(seedItem, (lastWrappedItem, currentItem) => 
{
  IWrappedItem wrappedItem = WrapItem(currentItem, lastWrappedItem);
  if (wrappedItem == null)
  {
    // 实现一些聪明的方法来跳过此无效项
    // 下一个项应该引用lastWrappedItem
  }
  return wrappedItem;
}));

在保持可观察模式的同时，如何实现这种行为而不向新集合返回null值？
是否有一个不同的方法我应该使用而不是"Scan"？

<details>
<summary>英文:</summary>

I have an IObservable of items with a timestamp.
I use the Scan method to wrap each item and add a reference to the last valid wrapped item which was received.

IObservable<IWrappedItem> wrappedItems =
allItems.Scan(seedItem,
(lastWrappedItem, currentItem) =>
WrapItem(currentItem, lastWrappedItem)));

This is the signature of `WrapItem`:

IWrappedItem WrapItem(Item currentItem, IWrappedItem lastItem);

We needed to change the `WrapItem` method so it skips invalid (not-null) items and returns null.
The seedItem will most probably be null, and the `WrapItem` method can handle it.

I need to update the way I use `Scan` with something like this:

IObservable<IWrappedItem> wrappedItems = allItems.Scan(seedItem, (lastWrappedItem, currentItem) =>
{
IWrappedItem wrappedItem = WrapItem(currentItem, lastWrappedItem);
if (wrappedItem == null)
{
// Do something clever to skip this invalid item
// Next item should have a reference to lastWrappedItem
}
return wrappedItem;
}));

How can I implement this behavior without returning null values to the new collection, while keeping the Observable pattern?
Is there a different method that I should use instead of "Scan"?

</details>


# 答案1
**得分**: 1

You should just be able to simply use the Where method https://learn.microsoft.com/en-us/dotnet/api/system.linq.enumerable.where?view=net-7.0

```c#
IObservable&lt;IWrappedItem&gt; wrappedItems = allItems.Where(item =&gt; item != null).Scan(seedItem, (lastWrappedItem, currentItem) =&gt; 
{
  IWrappedItem wrappedItem = WrapItem(currentItem, lastWrappedItem);
  if (wrappedItem == null)
  {
    // Do something clever to skip this invalid item
    // Next item should have a reference to lastWrappedItem
  }
  return wrappedItem;
}));

IObservable<IWrappedItem> wrappedItems = allItems.Where(item => item != null).Scan(seedItem, (lastWrappedItem, currentItem) => 
{
  IWrappedItem wrappedItem = WrapItem(currentItem, lastWrappedItem);
  if (wrappedItem == null)
  {
    // Do something clever to skip this invalid item
    // Next item should have a reference to lastWrappedItem
  }
  return wrappedItem;
}));

答案2

得分: 0

编辑：@TheodorZoulias - 感谢您的重要评论。
我进行了测试，并发现两个并行观察者不会同时运行，但第二个观察者会收到第一个观察者的lastWrappedItem对象，而不是null。因此，我像@Enigmativity建议的那样使用Defer来包装我的可观察对象，并且它能正常工作。

我找到了我的问题的答案，我实现了一个自定义通用的Scan方法，它接受WrapItem函数并忽略从中返回的null值。它使用Select和Where方法来实现Scan。

这是我的更新实现：

public static IObservable<TAccumulate> ScanObservableAndFilterNulls<TSource, TAccumulate>(this IObservable<TSource> items, TAccumulate seed, Func<TSource, TAccumulate, TAccumulate> wrapItemFunc)
{
  // 为了防止并发观察者的情况，需要在每个可观察对象中接收`previousDataItem`成员的新实例
  return Observable.Defer(() =>
  {
    // 在开始Scan实现之前使用种子
    TAccumulate lastWrappedItem = seed;
    // 实现自定义Scan方法
    return items.Select(item => wrapItemFunc(item, lastWrappedItem))
      .Where(wrappedItem =>
      {
        if (wrappedItem != null)
        {
          // 仅在包装项有效时更新lastWrappedItem
          lastWrappedItem = wrappedItem;
          return true;
        }
        // 跳过无效的包装项，但保留对最后一个有效项的引用
        return false;
      });
  });
}

这个方法可以这样使用：

public static IObservable<IWrappedItem> ScanAndWrapItems(IObservable<Item> allItems, IWrappedItem seedItem)
{
  return allItems.ScanObservableAndFilterNulls(seedItem, WrapItem);
}

我没有对新方法进行基准测试以评估性能损失，但我相信它会比原始的Scan方法慢...

public static IObservable<TAccumulate> ScanObservableAndFilterNulls<TSource, TAccumulate>(this IObservable<TSource> items, TAccumulate seed, Func<TSource, TAccumulate, TAccumulate> wrapItemFunc)
{
  // needed in order to protect from cases of concurrent observers
  // every observable will receive a new instance of the member `previousDataItem`
  return Observable.Defer(() =>
  {
    // use the seed before beginning the scan implementation
    TAccumulate lastWrappedItem = seed;
    // implement the custom Scan method
    return items.Select(item => wrapItemFunc(item, lastWrappedItem))
      .Where(wrappedItem =>
      {
        if (wrappedItem != null)
        {
          // update the lastWrappedItem only when the wrapped item is valid
          lastWrappedItem = wrappedItem;
          return true;
        }
        // skip invalid wrapped items, but keep the reference to the last valid item
        return false;
      });
  });
}

public static IObservable<IWrappedItem> ScanAndWrapItems(IObservable<Item> allItems, IWrappedItem seedItem)
{
  return allItems.ScanObservableAndFilterNulls(seedItem, WrapItem);
}