Is there a doAfterSubscribe equivalent in RxJava2?

I have an observable that streams responses to all requests. I'm wanting to create a filter of that observable when a request is made so that I can share the output with multiple subscribers. Following is some example code.

    PublishSubject<String> publishSubject = PublishSubject.create();
	Observable<String> fooObservable = publishSubject.filter(value -> value.startsWith("foo"))
			.doOnSubscribe(disposable -> {
				publishSubject.onNext("foobar");
			})
			.replay(1)
			.refCount();

	fooObservable.subscribe(val -> log.info("A val : <{}>", val));

I'm using a PublishSubject as my mock service because sometimes the service will return a response immediately.

What I'm finding is that because there is no current subscription when there is an immediate result my fooObservable is not being populated. i.e. I get no log output when I want to see :

A val : <foobar>

Note that I get the same result with this code :

PublishSubject<String> publishSubject = PublishSubject.create();
Observable<String> fooObservable = publishSubject.filter(value -> value.startsWith("foo"))
		.replay(1)
		.refCount();

publishSubject.onNext("foobar");
fooObservable.subscribe(val -> log.info("A val : <{}>", val));

So the issue is that the fooObservable doesn't subscribe to the PublishSubject until after it has been subscribed to,

Is there a way to run code immediately after the first subscription to the fooObservable?

Edit:
I thought about something like :

PublishSubject<String> publishSubject = PublishSubject.create();
BehaviorSubject<String> fooObservable = BehaviorSubject.create();
publishSubject.filter(value -> value.startsWith("foo")).subscribe(fooObservable);
publishSubject.onNext("foobar");
fooObservable.subscribe(val -> log.info("A val : <{}>", val));

But then I've got 2 subscriptions and I'm not sure how to clean up as the subscribe after the filter does not return a disposable.

Edit 2 : A description of the background task.

I have a third party service that my code needs to subscribe to. This service calls an onResponse method in my code with a parameter containing my original request and the response. The response may be updated by a new call to onResponse at any time.

I want to wrap to create a wrapper for this service that provides a method:

public Observable<Response> getObservable(Request req);

If the request matches one that is already subscribed then the observable should provide the most recent matching value immediately on subscription.

When there are no subscribers I need to unsubscribe from the service I'm wrapping.

I think publish() and connect() are what you need. Here you can read more about it.

In your case it's gonna be something like:

PublishSubject<String> publishSubject = PublishSubject.create();
Observable<String> fooObservable = publishSubject
    .filter(value -> value.startsWith("foo"))
    .publish();

fooObservable.subscribe(val -> log.info("A val : <{}>", val));
Observable<Object> o2 = fooObservable.map { new Object() }
Observable<Object> o3 = fooObservable.map { /* Something here*/ }

Disposable disposable = fooObservable.connect()

But remember to do disposable.dispose() to not have leak

EDIT

BehaviorSubject<String> publishSubject = BehaviorSubject.create();
Observable<String> fooObservable = publishSubject.filter(value ->         
value.startsWith("foo"));
fooObservable.subscribe(val -> log.info("A val : <{}>", val));
publishSubject.onNext("foobar");

I think the latest edit of the question helps explain what you're trying to do and this GitHub link probably has the answer you're looking for:
https://github.com/ReactiveX/RxJava/issues/4675

But I'll still share my test code. I ended up following the advice from the link above and using PublishSubject with .replay(1).refCount().

Let's say we have a third-party service interface:

interface ThirdPartyService
{
	void subscribe( Consumer<Integer> responseConsumer );
	
	void unsubscribe();
}

Next let's create a mock implementation that will invoke the consumer with a random int until it unsubscribes:

	// Mock service to emit a random integer once per second, no Rx:
	ThirdPartyService mockService = new ThirdPartyService() {
		
		Timer timer = new Timer();
		
		@Override
		public void subscribe( Consumer<Integer> responseConsumer )
		{
			System.out.println( "Subscribe" );
			Random random = new Random();
			
			TimerTask task = new TimerTask() {
		        @Override
				public void run()
		        {
		        	int i = random.nextInt( 10 );
		        	System.out.println( "Producing: " + i );
		        	responseConsumer.accept( i );
		        }
		    };
		    
		    timer.schedule( task, 1000, 1000 );
		}
		
		@Override
		public void unsubscribe()
		{
			System.out.println( "Unsubscribe" );
			timer.cancel();
		}
	};

Next, the actual Rx pipeline. Let's say I want to filter only odd integers and unsubscribe from service when there are no observers:

	// Wrap service in a PublishSubject:
	PublishSubject<Integer> subject = PublishSubject.create();
	mockService.subscribe( subject::onNext );
			
	// Create observable:
	Observable<Integer> observable = subject
			.doFinally( mockService::unsubscribe )
			.filter( i -> i % 2 == 1 )	// Include only odd integers
			.replay( 1 )				// Replay latest to new observers
			.refCount();

Finally, a manual test:

	// Subscribe to Observable:
	Disposable sub1 = observable.subscribe( i -> System.out.println( "sub1 got: " + i ));
	
	// Sleep:
	Thread.sleep( 3300 );
	
	// Create 2nd Subscriber:
	System.out.println( "adding sub2" );
	Disposable sub2 = observable.subscribe( i -> System.out.println( "sub2 got: " + i ));
	
	// Sleep:
	Thread.sleep( 3300 );
	
	// Dispose 2nd Subscriber:
	System.out.println( "disposing sub2" );
	sub2.dispose();
	
	// Sleep:
	Thread.sleep( 3300 );
	
	// Dispose 1st Subscriber:
	sub1.dispose();
	
	// Sleep:
	Thread.sleep( 3300 );

Output:

Subscribe
Producing: 1
sub1 got: 1
Producing: 8
Producing: 6
adding sub2
sub2 got: 1
Producing: 3
sub1 got: 3
sub2 got: 3
Producing: 7
sub1 got: 7
sub2 got: 7
Producing: 1
sub1 got: 1
sub2 got: 1
disposing sub2
Producing: 6
Producing: 7
sub1 got: 7
Producing: 1
sub1 got: 1
Unsubscribe