baku-toolkit/lib/src/main/java/io/v/rx/RxNamespace.java
/**
     * TODO(rosswang): This does not work properly if the name is relative and the namespace root is
     * changed during the life of the server. https://github.com/vanadium/issues/issues/1052
     *
     * @return an {@code Observable} of {@code MountEvent}s. Events including servers come from
     * polling and may or may not include an error. Events without servers are from
     * {@link Server#addName(String)} and may or may not include an error. If a server is already
     * mounted, a backdated mount event is included. The observable completes if and when the server
     * is unmounted.
     */
This does not work properly if the name is relative and the namespace root is
changed during the life of the server.

@return an  Observable of  MountEvents. Events including servers come from
polling and may or may not include an error. Events without servers are from
Server#addName(String) and may or may not include an error. If a server is already
mounted, a backdated mount event is included. The observable completes if and when the server
is unmounted.
This does not work properly if the name is relative and the namespace root is changed during the life of the server . @return an Observable of MountEvents . Events including servers come from polling and may or may not include an error . Events without servers are from Server#addName ( String ) and may or may not include an error . If a server is already mounted a backdated mount event is included . The observable completes if and when the server is unmounted .
public static Observable<MountEvent> mount(final Observable<Server> rxServer,
                                                          final String name) {
        return Observable.switchOnNext(
                rxServer.map(server -> {
                    /*
                    A note on thread safety. The initial status scan occurs on one thread, before
                    any concurrency contention. Subsequent modifications happen on the Rx io
                    scheduler, which by may be on different threads for each poll. Thus, not only
                    does the map need to be thread-safe, but also each mount record. The simplest
                    way to ensure this is just to lock map get/create and record access on a mutex,
                    which might as well be the map itself. This does perform some unnecessary
                    synchronization, but it all occurs on a worker thread on a low-fidelity loop, so
                    it's not worth a more sophisticated lock.
                    A fully correct minimal lock would include a thread-safe map and a read/write
                    lock for each record.
                     */
                    final Map<PublisherEntryKey, MountRecord> mountRecords = new HashMap<>();
                    final PublisherEntry[] mounts = server.getStatus().getPublisherStatus();
                    final List<MountEvent> alreadyMounted = new ArrayList<>(mounts.length);
                    for (final PublisherEntry status : mounts) {
                        if (status.getName().equals(name)) {
                            mountRecords.put(PublisherEntryKey.fromPublisherEntry(status),
                                    MountRecord.fromStatus(status));
                            if (isAlreadyMounted(status)) {
                                alreadyMounted.add(MountEvent.forStatus(true, status.getName(),
                                        status.getServer(), status.getLastMount(),
                                        status.getLastMountError() /* null */));
                            }
                        }
                    }
                    if (alreadyMounted.isEmpty()) {
                        try {
                            server.addName(name);
                        } catch (final VException e) {
                            return Observable.just(MountEvent.forAddNameFailure(name, e));
                        }
                        final MountEvent initial = MountEvent.forAddNameSuccess(name);
                        return RxPublisherState.index(
                                RxPublisherState.poll(server).map(state -> state.filter(
                                        status -> status.getName().equals(name))))
                                .flatMapIterable(Map::entrySet)
                                .concatMap(e -> processStatus(mountRecords,
                                        e.getKey(), e.getValue()))
                                .startWith(initial);
                    } else {
                        return Observable.from(alreadyMounted);
                    }
                }))
                .takeWhile(MountEvent::isMount);
    }
public static Observable < MountEvent > mount ( final Observable < Server > rxServer , final String name ) { return Observable . switchOnNext ( rxServer . map ( server -> { /*
                    A note on thread safety. The initial status scan occurs on one thread, before
                    any concurrency contention. Subsequent modifications happen on the Rx io
                    scheduler, which by may be on different threads for each poll. Thus, not only
                    does the map need to be thread-safe, but also each mount record. The simplest
                    way to ensure this is just to lock map get/create and record access on a mutex,
                    which might as well be the map itself. This does perform some unnecessary
                    synchronization, but it all occurs on a worker thread on a low-fidelity loop, so
                    it's not worth a more sophisticated lock.

                    A fully correct minimal lock would include a thread-safe map and a read/write
                    lock for each record.
                     */ final Map < PublisherEntryKey , MountRecord > mountRecords = new HashMap < > ( ) ; final PublisherEntry [ ] mounts = server . getStatus ( ) . getPublisherStatus ( ) ; final List < MountEvent > alreadyMounted = new ArrayList < > ( mounts . length ) ; for ( final PublisherEntry status : mounts ) { if ( status . getName ( ) . equals ( name ) ) { mountRecords . put ( PublisherEntryKey . fromPublisherEntry ( status ) , MountRecord . fromStatus ( status ) ) ; if ( isAlreadyMounted ( status ) ) { alreadyMounted . add ( MountEvent . forStatus ( true , status . getName ( ) , status . getServer ( ) , status . getLastMount ( ) , status . getLastMountError ( ) /* null */ ) ) ; } } } if ( alreadyMounted . isEmpty ( ) ) { try { server . addName ( name ) ; } catch ( final VException e ) { return Observable . just ( MountEvent . forAddNameFailure ( name , e ) ) ; } final MountEvent initial = MountEvent . forAddNameSuccess ( name ) ; return RxPublisherState . index ( RxPublisherState . poll ( server ) . map ( state -> state . filter ( status -> status . getName ( ) . equals ( name ) ) ) ) . flatMapIterable ( Map :: entrySet ) . concatMap ( e -> processStatus ( mountRecords , e . getKey ( ) , e . getValue ( ) ) ) . startWith ( initial ) ; } else { return Observable . from ( alreadyMounted ) ; } } ) ) . takeWhile ( MountEvent :: isMount ) ; }
TODO(rosswang): This does not work properly if the name is relative and the namespace root is
 changed during the life of the server.
TODO ( rosswang ) : This does not work properly if the name is relative and the namespace root is changed during the life of the server .