0

我正在尝试编写一个类，它将每隔 x 秒从数据库中查询一组数据。

问题是这个 observable 记住了它之前的状态，并且这个状态被用来确定 observable 上实际发送的内容。

例如我想要一个

IObservable<SomeEntity> AddedEntities { get; private set; }
IObservable<SomeEntity> ChangedEntities { get; private set; }
IObservable<SomeEntity> DeletedEntities { get; private set; }

我的问题是我一直在阅读http://www.introtorx.com/他们说使用主题是一个坏主意。相反，我应该使用 Observable.Create 方法。

我的问题是我只想查询数据库一次，然后将相关信息发送回正确的 observable，但我不希望每个订阅我的 observable 启动它自己的计时器，并且每次订阅每个 IObservable 时每 5 秒查询一次数据库。

我只制作了一个 observable，它返回一个新模型，并将更改为一组：

IObservable<EntityChangeSet> Changes {get; private set;}

public class EntityChangeSet
{
    Public IEnumerable<SomeEntity> Added {get; set;}
    Public IEnumerable<SomeEntity> Changed {get; set;}
    Public IEnumerable<SomeEntity> Deleted {get; set;}
}

我也对任何这样的解决方案持开放态度。

到目前为止，我所拥有的是：

public class IntervalChangeReader
{
    // My state between ticks.
    private IEnumerable<SomeEntity> knowEntities;

    // Reads the data from the db and uses knowEntities to determine adds, changes, 
    // and deletes which are exposed through properties.
    private DbReaderAndChangeChecker checker;

    IDisposeable timerCancel;

    public IntervalChangeReader(DbReaderAndChangeChecker checker)
    {
        this.checker = checker;
    }

    public IObservable<EntityChangeSet> Changes { get; private set; }

    public Start(int seconds)
    {
        this.timerCancel = Observable.Interval(new TimeSpan.FromSeconds(seconds)).Subscribe(
            x => 
            {
                var results = this.checker.Refresh(this.knownEntities);

                // Update this.knownEntities with results.


                // Inform produce a value on the observable ????
                // I could call .OnNext If I had a subject exposed for my Observable.
            }
    }

    public Stop()
    {
        this.timerCancel.Dispose();

        // Complete on all subscriptions?
        // If I were using subjects I could just call .OnComplete here.
    }
}

如何在不使用主题的情况下创建 Observable 以及如何通过它传递结果？我想只用一个计时器而不是每次订阅我的 observables 的计时器来完成这一切。

4

2 回答 2

1

我相信你所追求的是ConnectableObservable方法......玩弄这个（注意，在电话上，所以我的语法可能有点偏离）

public void Start(int seconds)
{
    // Every tick, generate a "changeset"
    var changeCentral =
        from tick in Observable.Interval(TimeSpan.FromSeconds(seconds))
        let results = this.checker.Refresh(this.knowEntities)
        select new EntityChangeSet(results);

    // Publish means one subscription for all "connecting" subscribers
    // RefCount means so long as one subscriber is subscribed, the subscription remains alive
    var connector = changeCentral.Publish().RefCount();
    Changes = connector;
}
于 2013-02-07T23:11:33.027 回答
0

我想我已经找到了解决问题的方法。

public class IntervalChangeReader
{
    // My state between ticks.
    private IEnumerable<SomeEntity> knowEntities;

    // Reads the data from the db and uses knowEntities to determine adds, changes, 
    // and deletes which are exposed through properties.
    private DbReaderAndChangeChecker checker;

    private EntityChangeSet lastChangeResult;

    public IntervalChangeReader(DbReaderAndChangeChecker checker)
    {
        this.checker = checker;

        // I like to use Do instead of just combining this all into Select to make the side effect obvious.
        this.Changes = Observable.Interval(TimeSpan.FromSeconds(seconds), scheduler).Do(
            x =>
                {
                    var changes = DbReaderAndChangeChecker.Refresh(this.knownXRefs);

                    // Make changes to knownXRefs which is my state.
                    this.knownXRefs.AddRange(changes.Added);
                    this.knownXRefs.RemoveAll(y => changes.Removed.Any(z => z.Id == y.Id));
                    this.knownXRefs.RemoveAll(y => changes.Changed.Any(z => z.Id == y.Id));
                    this.knownXRefs.AddRange(changes.Changed);

                    lastChangeResult = changes;
                }).Select(x => this.lastChangeResult);

            }

    public IObservable<EntityChangeSet> Changes { get; private set; }

}
于 2013-02-07T23:44:15.950 回答