A way to solve the "in-memory" vs "in-database" data modeling problem

DataUpdate.java

/*
 * Copyright (c) 2017 PlanGrid, Inc. All rights reserved.
 */

package com.plangrid.android.dmodel.mapper;

import java.util.concurrent.atomic.AtomicReference;

import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.subjects.BehaviorSubject;
import rx.subjects.Subject;

public class DataUpdate<T> {
    private final AtomicReference<Func1<T, T>> transformation = new AtomicReference<>(noOp());
    private final Subject<Func1<T, T>, Func1<T, T>> currentActionSubject = BehaviorSubject.create(transformation.get());

    private static <T> Func1<T, T> noOp() {
        return t -> t;
    }

    public T apply(final T data) {
        return transformation.get().call(data);
    }

    public void transform(final Action1<T> update) {
        final Func1<T, T> oldTransform = transformation.get();
        final Func1<T, T> compose = (final T t) -> {
            final T transform = oldTransform.call(t);
            update.call(transform);
            return transform;
        };

        if (!transformation.compareAndSet(oldTransform, compose)) {
            transform(update);
        } else {
            currentActionSubject.onNext(compose);
        }
    }

    public void transform(final Func1<T, T> update) {
        final Func1<T, T> oldTransform = transformation.get();
        final Func1<T, T> compose = (final T t) -> update.call(oldTransform.call(t));

        if (!transformation.compareAndSet(oldTransform, compose)) {
            transform(update);
        } else {
            currentActionSubject.onNext(compose);
        }
    }

    public Func1<T, T> reset() {
        return transformation.getAndSet(noOp());
    }

    private Observable<Func1<T, T>> observeUpdates() {
        return currentActionSubject.onBackpressureLatest();
    }

    public Observable<T> observableApply(final Observable<T> obs) {
        return Observable.combineLatest(observeUpdates(), obs, Func1::call);
    }
}