Reactive Extensions: клиент для условного api со стратегией Cache-Aside & Refresh-Ahead

[kriomant]

Сила реактивных расширений в том, что мы собираем pipeline из элементарных компонентов. В частности, кэширование можно представить в виде преобразования над Observable и в дальнейшем переиспользовать. В данном же случае загрузка данных из сети и кэширование смешаны, смысла в Observable почти ноль.

[return_true]

А можно подробнее про кэширование в виде Observable? Мне с ходу приходит в голову лишь асинхронная модель положили-достали.

[kriomant]

Disclaimer: чукча не C#-писатель и код не проверялся, так что где-то могут быть ошибки, важна идея.

Для начала разделим получение кармы и кэширование:

public static IObservable<KarmaModel> GetKarma(string userName) {
	return Observable.Create<KarmaModel>(observer =>
		Scheduler.Schedule(async () => {
			var karmaResponse = await HttpClient.Get(userName);
			if (!karmaResponse.IsSuccessful) {
				observer.OnError(karmaResponse.Exception);
			} else {
				observer.OnNext(karmaResponse.Data);
				observer.OnCompleted();
			}
		})
	);
}

public static IObservable<KarmaModel> Cache(this IObservable<KarmaModel> source, ICache cache, string userName) {
	return Observable.CreateWithDisposable<KarmaModel>(observer =>
		// Provide value from cache immediately if exists.
		KarmaMode karma = null;
		if (cache.HasCached(userName)) {
			karma = cache.GetCachedItem(userName);
			observer.OnNext(karma);
		}

		// Then just forward values to observer, caching last value.
		source.Subscribe(
			(karma) => {
				if (karma != null && updatedKarma.LastModified > karma.LastModified) {
					cache.Put(updatedKarma)
					observer.OnNext(karma)
				}
			},
			observer.OnError,
			observer.OnCompleted
		);
	);
}

теперь уберём ненужную зависимость кэширующего оператора от имени пользователя:

public interface ICache {
	bool Empty();
	KarmaModel Get();
	void Put(KarmaModel);
}

public static IObservable<KarmaModel> Cache(this IObservable<KarmaModel> source, ICache cache) {
	// Изменения очевидны
}

А что это за проверка на LastModified? Для этого есть DistinctUntilChanged с подходящим компаратором.

Итого получить что-то вроде:

var userName = "foo";
var userCache = globalCache.getByKey(userName);
var karma = new GetKarma(userName).Cache(userCache).DistinctUntilChanged(new KarmaLastModifiedComparator)

И теперь уже чётко видно, что куски можно обобщить и переиспользовать.

[return_true]

Ну, что я могу сказать? Снимаю шляпу — получилось достойно.

Не очень только понял в чём выгода с отдельным кэшем для каждого конкретного ключа?

[kriomant]

Потому что мы оперируем абстрактным потоком данных. В данном случае потоком кармы. Зачем кэширующему слою знать логин пользователя, зачем ему знать, по какому ключу и куда нужно складывать кэш? Тем более, что в случае обобщения будет неизвестно какой ключ, а то и вовсе ключа не будет (например, для единственного в своем роде потока данных). Пусть это знание про структуру кэша останется извне:

bla-bla.Cache(new ICache { Karma Get() { return cache.Get(userName); }; void Put(Karma karma) { cache.Put(userName, karma); ... })

Никто не заставляет использовать этот оператор непосредственно, можно на его основе построить кэширующий оператор, складывающий в Dictionary и его уже везде использовать. Важно лишь то, что нужно делать маленькие кирпичики и из них собирать стену, а не лить монолитом весь дом. Потому что из кирпичей ещё много чего полезного очень быстро можно собрать.

[return_true]

Эх, в C# так просто объект из интерфейса не создать.

В общем, мне ваш вариант понравился, хотя это некоторый оверкилл для демонстрации моей идеи. Но свою пользу и информацию к размышлению я получил, спасибо! Буду копать дальше.

[return_true]

Я ваш код на C# перевёл. Вот, что у меня получилось: RxHabraClient

.WithCache получился даже generic и не зависящий от ICache. Пожалуй так даже и лучше.

 public static IObservable<T> WithCache<T>(this IObservable<T> source, 
                                                Func<T> get, 
                                                Action<T> put) where T : class
        {
            return Observable.Create<T>(observer =>
            {
                var cached = get();
                if (cached != null)
                {
                    observer.OnNext(cached);
                }
                source.Subscribe(item =>
                {
                    put(item);
                    observer.OnNext(item);

                    observer.OnCompleted();
                }, observer.OnError);
                return Disposable.Empty;
            });
        }

И его использование:

.WithCache(() => Cache.GetCachedItem(userName), model => Cache.Put(model))

[Gangsta]

Вы определили общий интерфейс ICache как интерфейс, привязанный к модели (KarmaModel). В вашем случае правильно было бы писать IKarmaCache. Хотя лучше делать именно общий интерфейс. А модели наследовать от BaseModel.

Вот так бы я переписал ваш код:

public interface ICache
{
	bool HasCached(string key);
	BaseModel GetCachedItem(string key);
	void Put(BaseModel updatedModel);
}

public abstract BaseModel
{
	public string Key { get; set; }
}

public KarmaModel : BaseModel
{
	// ...
}

Дальше идет та же ошибка. Абстракция (IHttpClient) привязана к конкретной реализации (KarmaModel). Эта ошибка также проявляется в классе KarmaResponse, т.к. опять же, лучше сделать BaseResponse и от него унаследовать KarmaResponse.

Ну а так норм, неплохой подход.

Вы используете Xamarin? Какую библиотекую используете для MVVM?

[return_true]

Ох, надеюсь вы не всерьёз критикуете максимально упрощённый демо-код? :) Замечания-то правильные, продакшн-код был бы другой.

Что касается MVVM, то использовал разные: как самописные, так и стандартные (MVVMLight и надстройки над старым форком CalibrumMicro). Связку из Xamarin и MvvmCross использовал дома, а потом кончился триал.

Rx под Xamarin работают нормально. GitHub своё Android приложение пилит на Xamarin и там активно используют Rx и Rx-UI.