reactive programming - PublishSubject calls map for each present observer -

i'm using publishsubject along map operator:

@test public void testmapwithmultipleobservers() {      publishsubject<integer> subject = publishsubject.create();     func1 action = spy(new func1<integer, integer>() {         @override         public integer call(integer integer) {             return integer;         }     });     observable<integer> observable = subject.asobservable().map(action);      observable.subscribe(mock(observer.class));     observable.subscribe(mock(observer.class));      subject.onnext(1);      verify(action, times(2)).call(anyint());     // however, need times(1)  } 

the desired behaviour perform action after subject produces value. i've tried dooneach,doonnext, map , in each case action performed each present observer (for 100 observers action performed 100 times), while need perform per emission.

could suggest anything?

thanks.

the quickest option use share()

observable<integer> observable =      subject         .map(action)         .share(); 

you don't need asobservable() call. used return subject api , prevent caller casting subject. example:

observable<integer> getsubjectasobservable() {     return subject.asobservable(); }