使用流的请求/响应

Question

我有一个来自 RS-232 端口的输入流和一个使用 Rx 流到串行端口的命令队列。我已将代码简化如下：

void Init()
{
    SerialPort srl;

    ... // open serial port

    IObservable<string> obInput =
        Observable.FromEventPattern<
        SerialDataReceivedEventHandler,
        SerialDataReceivedEventArgs>
        (
            handler => srl.DataReceived += handler,
            handler => srl.DataReceived -= handler
        ).SelectMany(_ =>
        {
        List<string> ret;
        ... //extract messages
        return ret;
        }).Publish().Refcount();

    obCommandOk =
        obInput
        .Where(msg => msg == "OK" || msg == "KO");

    var sAction = new Subject<string>();
    var sCommandOk = new Subject<Tuple<string,bool>>();

    sAction
        .Do(srl.WriteLine)
        .Zip(obCommandOk, (cmd, result) =>
        {
        if (result == "OK")
            sCommandOk.OnNext(Tuple.Create(cmd, true))
        else
            sCommandOk.OnNext(Tuple.Create(cmd, false))
        });
}

async bool Command(string cmd)
{
    sAction.OnNext(cmd);

    return
        await sCommandOk
        .Where(t => t.Item1 == cmd)
        .Select(t => t.Item2)
        .FirstAsync();
}

有时会发生在 OnNext 之后结果已经被推送到 sCommandOk 的情况，所以我丢失了它。

你能建议我一个更好的方法来避免丢失响应吗？

Answer 1

您的 Command 方法中有竞争条件。 您推送操作，然后订阅结果。 如果结果很快，那么你就会失去它。

这里的一个小改动可以缓解这种情况：

async Task<bool> Command(string cmd)
{
    var result = sCommandOk
        .Where(t => t.Item1 == cmd)
        .Select(t => t.Item2)
        .FirstAsync();

    sAction.OnNext(cmd);

    return await result;
}

我认为您可以通过将sCommandOk 主题全部删除来进一步优化

async Task<bool> Command(string cmd)
{
    var result = obInput
        .Where(t => t.Item1 == cmd)
        .Select(t => t.Item2)
        .FirstAsync();

    sAction.OnNext(cmd);

    return await result;
}