英文:
Is there a cleaner way adapt a standard observer/listener service using RxJava?
问题
static class MockServiceRXAdapterImpl1 implements MockServiceRXAdapter {
PublishSubject<MockResponse> mockResponseObservable = PublishSubject.create();
MockService mockService = new MockService(mockResponse -> mockResponseObservable.onNext(mockResponse));
final ConcurrentMap<String, Observable<String>> subscriptionMap = new ConcurrentHashMap<>();
public Observable<String> getObservable(String id) {
return Observable.using(
() -> subscriptionMap.computeIfAbsent(
id,
key -> mockResponseObservable.filter(mockResponse -> mockResponse.id.equals(id))
.doOnSubscribe(disposable -> mockService.subscribe(id))
.doOnDispose(() -> {
mockService.unsubscribe(id);
subscriptionMap.remove(id);
})
.map(mockResponse -> mockResponse.value)
.replay(1)
.refCount()
),
observable -> observable,
observable -> {}
);
}
}
英文:
I've been tinkering with wrapping an old style listener interface using RxJava. What i've come up with seems to work, but the usage of Observable.using feels a bit awkward.
The requirements are:
- Only one subscription per id to the underlying service.
- The latest value for a given id should be cached and served to new subscribers.
- We must unsubscribe from the underlying service if nothing is listening to an id.
Is there a better way? The following is what I've got.
static class MockServiceRXAdapterImpl1 implements MockServiceRXAdapter {
PublishSubject<MockResponse> mockResponseObservable = PublishSubject.create();
MockService mockService = new MockService(mockResponse -> mockResponseObservable.onNext(mockResponse));
final ConcurrentMap<String, Observable<String>> subscriptionMap = new ConcurrentHashMap<>();
public Observable<String> getObservable(String id) {
return Observable.using(() -> subscriptionMap.computeIfAbsent(
id,
key -> mockResponseObservable.filter(mockResponse -> mockResponse.id.equals(id))
.doOnSubscribe(disposable -> mockService.subscribe(id))
.doOnDispose(() -> {
mockService.unsubscribe(id);
subscriptionMap.remove(id);
})
.map(mockResponse -> mockResponse.value)
.replay(1)
.refCount()),
observable -> observable,
observable -> {
}
);
}
}
答案1
得分: 1
你可以使用 Observable.create。
代码可能看起来像这样：
final Map<String, Observable<String>> subscriptionMap = new HashMap<>();
MockService mockService = new MockService();
public Observable<String> getObservable(String id) {
log.info("looking for root observable");
if (subscriptionMap.containsKey(id)) {
log.info("found root observable");
return subscriptionMap.get(id);
} else {
synchronized (subscriptionMap) {
if (!subscriptionMap.containsKey(id)) {
log.info("creating new root observable");
final Observable<String> responseObservable = Observable.create(emitter -> {
MockServiceListener listener = emitter::onNext;
mockService.addListener(listener);
emitter.setCancellable(() -> {
mockServices.removeListener(listener);
mockService.unsubscribe(id);
synchronized (subscriptionMap) {
subscriptionMap.remove(id);
}
});
mockService.subscribe(id);
})
.filter(mockResponse -> mockResponse.id.equals(id))
.map(mockResponse -> mockResponse.value)
.replay(1)
.refCount();
subscriptionMap.put(id, responseObservable);
} else {
log.info("Another thread created the observable for us");
}
return subscriptionMap.get(id);
}
}
}
英文:
You may use Observable.create
So code may look like this
final Map<String, Observable<String>> subscriptionMap = new HashMap<>();
MockService mockService = new MockService();
public Observable<String> getObservable(String id) {
log.info("looking for root observable");
if (subscriptionMap.containsKey(id)) {
log.info("found root observable");
return subscriptionMap.get(id);
} else {
synchronized (subscriptionMap) {
if (!subscriptionMap.containsKey(id)) {
log.info("creating new root observable");
final Observable<String> responseObservable = Observable.create(emitter -> {
MockServiceListener listener = emitter::onNext;
mockService.addListener(listener);
emitter.setCancellable(() -> {
mockServices.removeListener(listener);
mockService.unsubscribe(id);
synchronized (subscriptionMap) {
subscriptionMap.remove(id);
}
});
mockService.subscribe(id);
})
.filter(mockResponse -> mockResponse.id.equals(id))
.map(mockResponse -> mockResponse.value)
.replay(1)
.refCount();
subscriptionMap.put(id, responseObservable);
} else {
log.info("Another thread created the observable for us");
}
return subscriptionMap.get(id);
}
}
}
答案2
得分: 0
以下是翻译好的内容：
我认为我已经使用.groupBy(...)使其工作。
在我的情况下，Response.getValue() 返回一个int，但是你可以理解：
class Adapter
{
Subject<Response> msgSubject;
ThirdPartyService service;
Map<String, Observable<Integer>> observables;
Observable<GroupedObservable<String, Response>> groupedObservables;
public Adapter()
{
msgSubject = PublishSubject.<Response>create().toSerialized();
service = new MockThirdPartyService(msgSubject::onNext);
groupedObservables = msgSubject.groupBy(Response::getId);
observables = Collections.synchronizedMap(new HashMap<>());
}
public Observable<Integer> getObservable(String id)
{
return observables.computeIfAbsent(id, this::doCreateObservable);
}
private Observable<Integer> doCreateObservable(String id)
{
service.subscribe(id);
return groupedObservables
.filter(group -> group.getKey().equals(id))
.doOnDispose(() -> {
synchronized (observables)
{
service.unsubscribe(id);
observables.remove(id);
}
})
.concatMap(Functions.identity())
.map(Response::getValue)
.replay(1)
.refCount();
}
}
英文:
I think I've gotten it to work using .groupBy(...).
In my case Response.getValue() returns an int, but you get the idea:
class Adapter
{
Subject<Response> msgSubject;
ThirdPartyService service;
Map<String, Observable<Integer>> observables;
Observable<GroupedObservable<String, Response>> groupedObservables;
public Adapter()
{
msgSubject = PublishSubject.<Response>create().toSerialized();
service = new MockThirdPartyService( msgSubject::onNext );
groupedObservables = msgSubject.groupBy( Response::getId );
observables = Collections.synchronizedMap( new HashMap<>() );
}
public Observable<Integer> getObservable( String id )
{
return observables.computeIfAbsent( id, this::doCreateObservable );
}
private Observable<Integer> doCreateObservable( String id )
{
service.subscribe( id );
return groupedObservables
.filter( group -> group.getKey().equals( id ))
.doOnDispose( () -> {
synchronized ( observables )
{
service.unsubscribe( id );
observables.remove( id );
}
} )
.concatMap( Functions.identity() )
.map( Response::getValue )
.replay( 1 )
.refCount();
}
}
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论