Сейчас у меня код не использующий Rx выглядит так:

interface ICallback
{
 Run(string argument);
}

class CallbackManager
{
  // Id -> list of subscribers
  Dictionary<string, List<ICallback>> callbacks;

  void Process(string id, string argument)
  {
    foreach(var c in callbacks[id]) c.Run(argument);
  }
}

Как грамотно сделать выбор по словарю в Rx ?
Есть конечно Where, но тогда поиск ключа будет линейным.
Можно сделать Buffer/Window + GroupBy, но как-то выглядит неправильным.

Спасибо.

Здравствуйте, _NN_, Вы писали:

_NN>Как грамотно сделать выбор по словарю в Rx ?

Первое пришедшее в голову — .Select()/.SelectMany(),

 observable.SelectMany(id => dict[id]);

если не все id есть в словаре — вставить .Where() или переписать SelectMany на (псевдокод):

 id => dict.TryGetValue(id, out callbacks)? callbacks: Enumerable.Empty<TCallback>();

P.S. Проверка словаря на наличие ключа — O(1), линейного поиска не будет.

Здравствуйте, Sinix, Вы писали:

S>P.S. Проверка словаря на наличие ключа — O(1), линейного поиска не будет.

Т.е. от кода работы со словарем не уйти ?
Я думал может в Rx есть какой-то готовый маршрутизатор.

Здравствуйте, _NN_, Вы писали:

_NN>Как грамотно сделать выбор по словарю в Rx ?
_NN>Есть конечно Where, но тогда поиск ключа будет линейным.
_NN>Можно сделать Buffer/Window + GroupBy, но как-то выглядит неправильным.

_NN>Спасибо.

Все подписчики будут реагировать на каждое новое событие в потоке, другое дело что код этого подписчика может быть не вызван.
Например если ты добавишь .Where(actionId => actionId == subscriberId), то каждому подписчику придётся проверять, ему ли адресовано данное сообщение.
Неважно какие операторы ты добавишь между потоком и Subscribe — с каждым новым подписчиком у тебя будет возрастать количество проверок.

К сожалению, даже используя RX, тут без словаря не обойтись. Выглядеть будет примерно так:

public static void Main()
{
    GetStreamById(1).Subscribe(a => Console.WriteLine("Subscriber with id = 1 called. Argument = " + a));
    GetStreamById(1).Subscribe(a => Console.WriteLine("Subscriber with id = 1 called. Argument = " + a));
    GetStreamById(3).Subscribe(a => Console.WriteLine("Subscriber with id = 3 called. Argument = " + a));
    GetStreamById(4).Subscribe(a => Console.WriteLine("Subscriber with id = 4 called. Argument = " + a));
    
    _stream.OnNext(Tuple.Create(1, "a"));
    _stream.OnNext(Tuple.Create(1, "b"));
    _stream.OnNext(Tuple.Create(4, "a"));
    _stream.OnNext(Tuple.Create(3, "b"));
}

static Subject<Tuple<int, string>> _stream = new Subject<Tuple<int, string>>();
static Dictionary<int, IObservable<string>> _streamsById = new Dictionary<int, IObservable<string>>();

public static IObservable<string> GetStreamById(int id)
{
    IObservable<string> stream;
    
    if (_streamsById.TryGetValue(id, out stream))
        return stream;

    var newStream = _stream.Where(t => t.Item1 == id)
                           .Select(t => t.Item2)
                           .Publish();
    
    newStream.Connect();
    
    return _streamsById[id] = newStream;
}

Результат:

Subscriber with id = 1 called. Argument = a
Subscriber with id = 1 called. Argument = a
Subscriber with id = 1 called. Argument = b
Subscriber with id = 1 called. Argument = b
Subscriber with id = 4 called. Argument = a
Subscriber with id = 3 called. Argument = b

Здравствуйте, ionoy, Вы писали:

Забыл дать ссылку на фиддл: https://dotnetfiddle.net/FF6yYi

Здравствуйте, _NN_, Вы писали:

_NN>Здравствуйте, Sinix, Вы писали:

S>>P.S. Проверка словаря на наличие ключа — O(1), линейного поиска не будет.

_NN>Т.е. от кода работы со словарем не уйти ?
_NN>Я думал может в Rx есть какой-то готовый маршрутизатор.

GroupBy

Здравствуйте, gandjustas, Вы писали:

_NN>>Т.е. от кода работы со словарем не уйти ?
_NN>>Я думал может в Rx есть какой-то готовый маршрутизатор.

G>GroupBy

Имхо, у него будет та же линейная сложность.

Здравствуйте, Аноним, Вы писали:

А>Здравствуйте, gandjustas, Вы писали:

_NN>>>Т.е. от кода работы со словарем не уйти ?
_NN>>>Я думал может в Rx есть какой-то готовый маршрутизатор.

G>>GroupBy

А>Имхо, у него будет та же линейная сложность.
С чего ты взял?
Внутри используется Disctionary для группировки, получение элемента из Dictonary это O(1).

Здравствуйте, gandjustas, Вы писали:

G>Здравствуйте, Аноним, Вы писали:

А>>Здравствуйте, gandjustas, Вы писали:

_NN>>>>Т.е. от кода работы со словарем не уйти ?
_NN>>>>Я думал может в Rx есть какой-то готовый маршрутизатор.

G>>>GroupBy

А>>Имхо, у него будет та же линейная сложность.
G>С чего ты взял?
G>Внутри используется Disctionary для группировки, получение элемента из Dictonary это O(1).

Да, только для каждого подписчика будет свой Dictionary.

Здравствуйте, Аноним, Вы писали:

А>Здравствуйте, gandjustas, Вы писали:

G>>Здравствуйте, Аноним, Вы писали:

А>>>Здравствуйте, gandjustas, Вы писали:

_NN>>>>>Т.е. от кода работы со словарем не уйти ?
_NN>>>>>Я думал может в Rx есть какой-то готовый маршрутизатор.

G>>>>GroupBy

А>>>Имхо, у него будет та же линейная сложность.
G>>С чего ты взял?
G>>Внутри используется Disctionary для группировки, получение элемента из Dictonary это O(1).

А>Да, только для каждого подписчика будет свой Dictionary.

Если вызовешь несколько раз Subscribe для GroupBy, то да, но зачем это делать? Смысл GroupBy в том что ты подписываешься на результат GroupBy, а внутри обсервера подписываешься на каждую группу.

Здравствуйте, gandjustas, Вы писали:

G>Если вызовешь несколько раз Subscribe для GroupBy, то да, но зачем это делать? Смысл GroupBy в том что ты подписываешься на результат GroupBy, а внутри обсервера подписываешься на каждую группу.
А как будет выглядеть окончательный код для этой задачи?

Здравствуйте, Аноним, Вы писали:

А>Здравствуйте, gandjustas, Вы писали:

G>>Если вызовешь несколько раз Subscribe для GroupBy, то да, но зачем это делать? Смысл GroupBy в том что ты подписываешься на результат GroupBy, а внутри обсервера подписываешься на каждую группу.
А>А как будет выглядеть окончательный код для этой задачи?

https://dotnetfiddle.net/ieN8OJ

Здравствуйте, gandjustas, Вы писали:

G>Здравствуйте, Аноним, Вы писали:

А>>Здравствуйте, gandjustas, Вы писали:

G>>>Если вызовешь несколько раз Subscribe для GroupBy, то да, но зачем это делать? Смысл GroupBy в том что ты подписываешься на результат GroupBy, а внутри обсервера подписываешься на каждую группу.
А>>А как будет выглядеть окончательный код для этой задачи?

G>https://dotnetfiddle.net/ieN8OJ

1. Для подписки нужен отдельный метод
2. Отписка ещё сложнее
3. У подписчиков нет доступа к IObservable

Вроде используем Rx, а преимуществ не получаем. Конечно, если подходить к задаче строго и переписывать её один в один, то решение работает. Но толку от него мало.

Здравствуйте, ionoy, Вы писали:

I>Здравствуйте, gandjustas, Вы писали:

G>>Здравствуйте, Аноним, Вы писали:

А>>>Здравствуйте, gandjustas, Вы писали:

G>>>>Если вызовешь несколько раз Subscribe для GroupBy, то да, но зачем это делать? Смысл GroupBy в том что ты подписываешься на результат GroupBy, а внутри обсервера подписываешься на каждую группу.
А>>>А как будет выглядеть окончательный код для этой задачи?

G>>https://dotnetfiddle.net/ieN8OJ

I>1. Для подписки нужен отдельный метод
I>2. Отписка ещё сложнее
I>3. У подписчиков нет доступа к IObservable
О чем это?

I>Вроде используем Rx, а преимуществ не получаем. Конечно, если подходить к задаче строго и переписывать её один в один, то решение работает. Но толку от него мало.
Вы пытаетесь натянть Rx на свой код. Это, увы, не работает. RX, как и любое средство композиции частей программы, плохо дружит с другими средствами композиции.

Здравствуйте, gandjustas, Вы писали:

G>>>https://dotnetfiddle.net/ieN8OJ

I>>1. Для подписки нужен отдельный метод
I>>2. Отписка ещё сложнее
I>>3. У подписчиков нет доступа к IObservable
G>О чем это?

О том, что в моём примере каждый клиент получает IObservable<Result> и может делать с ним что хочет (Where, SelectMany и т.д).
Какой смысл использовать Rx, если нет возможности изменять последовательность?

I>>Вроде используем Rx, а преимуществ не получаем. Конечно, если подходить к задаче строго и переписывать её один в один, то решение работает. Но толку от него мало.
G>Вы пытаетесь натянть Rx на свой код. Это, увы, не работает. RX, как и любое средство композиции частей программы, плохо дружит с другими средствами композиции.
Зато можно адаптировать свой код к Rx и потом пользоваться результатом из других частей программы. Там где были простые callback'и, теперь нормальные IObservable со всеми вытекающими последствиями.

Здравствуйте, ionoy, Вы писали:

I>Здравствуйте, gandjustas, Вы писали:

G>>>>https://dotnetfiddle.net/ieN8OJ

I>>>1. Для подписки нужен отдельный метод
I>>>2. Отписка ещё сложнее
I>>>3. У подписчиков нет доступа к IObservable
G>>О чем это?

I>О том, что в моём примере каждый клиент получает IObservable<Result> и может делать с ним что хочет (Where, SelectMany и т.д).
I>Какой смысл использовать Rx, если нет возможности изменять последовательность?
А с чего ты взял что нельзя?

var d = _stream.GroupBy(t => t.Item1, t=> t.Item2).Subscribe(g => {
    // g:IObservable<string>  - делай с ним что хочешь.
});

Скорее всего твоя ментальная модель не совпадает с ментальной моделью Rx. Это не проблема Rx если что.

Здравствуйте, gandjustas, Вы писали:

G>Здравствуйте, ionoy, Вы писали:

I>>Здравствуйте, gandjustas, Вы писали:

G>>>>>https://dotnetfiddle.net/ieN8OJ

I>>>>1. Для подписки нужен отдельный метод
I>>>>2. Отписка ещё сложнее
I>>>>3. У подписчиков нет доступа к IObservable
G>>>О чем это?

G>А с чего ты взял что нельзя?

G>

G>var d = _stream.GroupBy(t => t.Item1, t=> t.Item2).Subscribe(g => {
G>    // g:IObservable<string>  - делай с ним что хочешь.
G>});
G>

Ты реально задачу не понимаешь?
1. Подписчик может быть добавлен в любой момент и при этом иметь возможность работать с IObservable так как именно ему удобно, а не всем остальным подписчикам.
2. Иметь отдельные методы для подписки и отписки — это конечно очень в рамках ментальной модели Rx.

G>>Скорее всего твоя ментальная модель не совпадает с ментальной моделью Rx. Это не проблема Rx если что.
Скорее всего ты сел в лужу и не хочешь этого признавать.

Здравствуйте, ionoy, Вы писали:

G>>А с чего ты взял что нельзя?

G>>

G>>var d = _stream.GroupBy(t => t.Item1, t=> t.Item2).Subscribe(g => {
G>>    // g:IObservable<string>  - делай с ним что хочешь.
G>>});
G>>

I>Ты реально задачу не понимаешь?
Телепатия не сработала.

I>1. Подписчик может быть добавлен в любой момент и при этом иметь возможность работать с IObservable так как именно ему удобно, а не всем остальным подписчикам.
https://dotnetfiddle.net/KVNbXj

I>2. Иметь отдельные методы для подписки и отписки — это конечно очень в рамках ментальной модели Rx.
Иметь отдельные методы для подписки и отписки для последовательности, которой не существует — за рамками ментальной модели.

А чтобы получить реальную последовательность тебе надо применить фильтр, получится O(N) от количества подписчиков.

G>>>Скорее всего твоя ментальная модель не совпадает с ментальной моделью Rx. Это не проблема Rx если что.
I>Скорее всего ты сел в лужу и не хочешь этого признавать.
Ну если тебе проще в это верить, то я ничем помочь не могу.

Здравствуйте, gandjustas, Вы писали:

I>>Ты реально задачу не понимаешь?
G>Телепатия не сработала.
То ли ты не тот код мне дал, то ли она сработала.
Преимущество Rx в том, что подписчик имеет дело с IObservable<T>, а не T. В твоём примере мы один раз подписываемся на на каждую из сгруппированных последовательностей. После этого уже невозможно добавить нового подписчика, который хочет взять только один элемент например (Take(1)). Всё это придётся описывать императивным кодом внутри самого подписчика.

I>>1. Подписчик может быть добавлен в любой момент и при этом иметь возможность работать с IObservable так как именно ему удобно, а не всем остальным подписчикам.
G>https://dotnetfiddle.net/KVNbXj
Ты это уже приводил, все недостатки на месте

I>>2. Иметь отдельные методы для подписки и отписки — это конечно очень в рамках ментальной модели Rx.
G>Иметь отдельные методы для подписки и отписки для последовательности, которой не существует — за рамками ментальной модели.
G>А чтобы получить реальную последовательность тебе надо применить фильтр, получится O(N) от количества подписчиков.
У меня конечно не совсем корректный пример получился, ты прав. Для того, чтобы избежать O(N) там придётся заменить Where на GroupBy и добавить Subject'ы, чтобы можно было подписываться до прихода первых элементов. Тогда будет реальная последовательность.

G>>>>Скорее всего твоя ментальная модель не совпадает с ментальной моделью Rx. Это не проблема Rx если что.
I>>Скорее всего ты сел в лужу и не хочешь этого признавать.
G>Ну если тебе проще в это верить, то я ничем помочь не могу.
Я бы мог ответить то же самое.

Здравствуйте, ionoy, Вы писали:

I>Забыл дать ссылку на фиддл: https://dotnetfiddle.net/FF6yYi

Зачем тут Publish и Connect? Кто удалит источники когда они станут не нужны?
Тут надо смотреть на Observable.RefCount, Observable.Defer и Observable.Create