i need to use rxjava in my business in layered architecture now my service impl is like this 

package sap.ass01.layered.services.impl;

import io.vertx.core.eventbus.EventBus;
import sap.ass01.layered.domain.model.EBike;
import sap.ass01.layered.domain.model.Ride;
import sap.ass01.layered.domain.model.RideEntry;
import sap.ass01.layered.domain.model.User;
import sap.ass01.layered.services.dto.EBikeDTO;
import sap.ass01.layered.services.dto.RideDTO;
import sap.ass01.layered.services.dto.UserDTO;
import sap.ass01.layered.services.simulation.EBikeObservable;
import sap.ass01.layered.services.simulation.RideSimulation;
import sap.ass01.layered.services.Services.AdminService;
import sap.ass01.layered.services.Services.LoginService;
import sap.ass01.layered.services.Services.UserService;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

public class ServiceImpl implements AdminService, LoginService, UserService {
    private final ConcurrentHashMap<String, EBike> bikes = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, User> users = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, RideEntry> rideEntries = new ConcurrentHashMap<>();


    private final EventBus eventBus;
    private final EBikeObservable ebikeObservable;

    public ServiceImpl(EventBus eventBus) {
        this.eventBus = eventBus;
        this.ebikeObservable = new EBikeObservable(eventBus, "bike.updates.all", "bike.updates.available");
    }

    @Override
    public CompletableFuture<EBikeDTO> createEBike(String bikeId, int x, int y) {
        return CompletableFuture.supplyAsync(() -> {
            EBike bike = new EBike(bikeId, x, y);
            bikes.put(bikeId, bike);
            notifyAllBikesObservers();
            return new EBikeDTO(bike.getId(), bike.getLocation().x(), bike.getLocation().y(), bike.getBatteryLevel(), bike.getState().name());
        });
    }

    @Override
    public CompletableFuture<Void> signIn(String name, Boolean admin) {
        return CompletableFuture.runAsync(() -> {
            User user = new User(name, admin ? User.UserType.ADMIN : User.UserType.USER);
            users.put(name, user);
        });
    }

    @Override
    public CompletableFuture<UserDTO> logIn(String name) {
        return CompletableFuture.supplyAsync(() -> {
            User user = users.get(name);
            return new UserDTO(user.getId(), user.getCredit(), user.getPermission().name());
        });
    }

    @Override
    public CompletableFuture<Void> startRide(String userId, String bikeId) {
        return CompletableFuture.runAsync(() -> {
            User user = users.get(userId);
            EBike bike = bikes.get(bikeId);
            Ride ride = new Ride(UUID.randomUUID().toString(), user, bike);
            ride.start();

            String rideUpdateAddress = "ride.updates." + ride.getId();
            RideSimulation rideSimulation = new RideSimulation(ride, user, eventBus, rideUpdateAddress);
            rideEntries.put(ride.getId(), new RideEntry(ride, rideSimulation));
            rideSimulation.start();
        });
    }

    @Override
    public CompletableFuture<Void> endRide(String userId, String bikeId) {
        return CompletableFuture.runAsync(() -> {
            RideEntry rideEntry = rideEntries.values().stream()
                    .filter(entry -> entry.ride().getUser().getId().equals(userId) && entry.ride().getEBike().getId().equals(bikeId))
                    .findFirst()
                    .orElseThrow(() -> new IllegalArgumentException("Ride not found"));
            rideEntry.ride().end();
            rideEntry.simulation().stopSimulation();
            rideEntries.remove(rideEntry.ride().getId());

            notifyAvailableBikesObservers();
        });
    }

    public CompletableFuture<UserDTO> getUser(String id) {
        return CompletableFuture.supplyAsync(() -> {
            User user = users.get(id);
            return new UserDTO(user.getId(), user.getCredit(), user.getPermission().name());
        });
    }

    @Override
    public CompletableFuture<Observable<Collection<EBikeDTO>>> getAllEBikes() {
        return CompletableFuture.supplyAsync(() -> ebikeObservable.getAllBikesObservable());
    }

    @Override
    public CompletableFuture<Observable<Collection<EBikeDTO>>> getAvailableEBikes() {
        return CompletableFuture.supplyAsync(() -> ebikeObservable.getAvailableBikesObservable());
    }

    private void notifyAllBikesObservers() {
        Collection<EBikeDTO> bikeDTOs = new ArrayList<>();
        for (EBike bike : bikes.values()) {
            bikeDTOs.add(new EBikeDTO(bike.getId(), bike.getLocation().x(), bike.getLocation().y(), bike.getBatteryLevel(), bike.getState().name()));
        }
        ebikeObservable.notifyAllBikesObservers(bikeDTOs);
    }

    private void notifyAvailableBikesObservers() {
        Collection<EBikeDTO> bikeDTOs = new ArrayList<>();
        for (EBike bike : bikes.values()) {
            if (bike.getState() == EBike.EBikeState.AVAILABLE) {
                bikeDTOs.add(new EBikeDTO(bike.getId(), bike.getLocation().x(), bike.getLocation().y(), bike.getBatteryLevel(), bike.getState().name()));
            }
        }
        ebikeObservable.notifyAvailableBikesObservers(bikeDTOs);
    }
}
 i just done these interfaces 

package sap.ass01.layered.services.Services;

import sap.ass01.layered.services.dto.EBikeDTO;
import io.reactivex.Single;
import io.reactivex.Observable;
import java.util.Collection;

public interface AdminService {
    /**
     * Creates a new E-Bike with the given ID and initial location.
     * @param bikeId Unique identifier for the bike.
     * @param x Initial X-coordinate.
     * @param y Initial Y-coordinate.
     * @return Single emitting the created EBikeDTO or an error.
     */
    Single<EBikeDTO> createEBike(String bikeId, double x, double y);

    /**
     * Observes all E-Bikes in real-time.
     * Admins can subscribe to receive updates about all bikes.
     * @return Observable emitting collections of EBikeDTO.
     */
    Observable<Collection<EBikeDTO>> observeAllBikes();
}

package sap.ass01.layered.services.Services;

import sap.ass01.layered.services.dto.UserDTO;
import io.reactivex.Completable;
import io.reactivex.Single;

public interface LoginService {
    /**
     * Registers a new user.
     * @param name Username.
     * @param isAdmin True if the user is an admin, false otherwise.
     * @return Completable indicating success or failure.
     */
    Completable signUp(String name, boolean isAdmin);

    /**
     * Authenticates a user by name.
     * @param name Username.
     * @return Single emitting UserDTO on success or an error.
     */
    Single<UserDTO> logIn(String name);
}

package sap.ass01.layered.services.Services;

import sap.ass01.layered.services.dto.EBikeDTO;
import sap.ass01.layered.services.dto.RideDTO;
import io.reactivex.Observable;
import io.reactivex.Completable;
import java.util.Collection;

public interface UserService {
    /**
     * Observes available E-Bikes in real-time.
     * Users can subscribe to receive updates about available bikes.
     * @return Observable emitting collections of EBikeDTO.
     */
    Observable<Collection<EBikeDTO>> observeAvailableBikes();

    /**
     * Starts a ride for a user on a specific bike.
     * @param userId ID of the user.
     * @param rideId Unique ID for the ride.
     * @param bikeId ID of the bike.
     * @return Completable indicating success or failure.
     */
    Completable startRide(String userId, String rideId, String bikeId);

    /**
     * Ends a ride for a user on a specific bike.
     * @param userId ID of the user.
     * @param rideId ID of the ride.
     * @param bikeId ID of the bike.
     * @return Completable indicating success or failure.
     */
    Completable endRide(String userId, String rideId, String bikeId);

    /**
     * Observes ride updates specific to a user's ride.
     * @param userId ID of the user.
     * @param rideId ID of the ride.
     * @return Observable emitting RideDTO updates.
     */
    Observable<RideDTO> observeRide(String userId, String rideId);
}


in presentation layer i can have two type of users an normal user that can see oblu byke that are avialable and ride a bike so the use have to see at runtime the bike only available so get a ubtae if one bike change state and do a ride so he have to se his bike riding i can have multiple user view active so every use can see only his ride and admin user that can see all the byke also the riding one mobving 

to achieve this i use a simulaito 
package sap.ass01.layered.services.simulation;

import io.vertx.core.eventbus.EventBus;
import sap.ass01.layered.domain.model.Ride;
import sap.ass01.layered.domain.model.User;
import sap.ass01.layered.domain.model.V2d;
import sap.ass01.layered.services.dto.RideDTO;

public class RideSimulation extends Thread {

    private Ride ride;
    private User user;
    private volatile boolean stopped;
    private EventBus eventBus;
    private String rideUpdateAddress;

    public RideSimulation(Ride ride, User user, EventBus eventBus, String rideUpdateAddress) {
        this.ride = ride;
        this.user = user;
        this.eventBus = eventBus;
        this.rideUpdateAddress = rideUpdateAddress;
        stopped = false;
    }

    public void run() {
        var b = ride.getEBike();
        b.updateSpeed(1);

        var lastTimeDecreasedCredit = System.currentTimeMillis();
        user.decreaseCredit(1);

        var lastTimeChangedDir = System.currentTimeMillis();

        while (!stopped) {
            var l = b.getLocation();
            var d = b.getDirection();
            var s = b.getSpeed();
            b.updateLocation(l.sum(d.mul(s)));
            l = b.getLocation();
            if (l.x() > 200 || l.x() < -200) {
                b.updateDirection(new V2d(-d.x(), d.y()));
            }
            if (l.y() > 200 || l.y() < -200) {
                b.updateDirection(new V2d(d.x(), -d.y()));
            }

            var elapsedTimeSinceLastChangeDir = System.currentTimeMillis() - lastTimeChangedDir;
            if (elapsedTimeSinceLastChangeDir > 500) {
                b.updateDirection(new V2d(Math.random() - 0.5, Math.random() - 0.5));
                lastTimeChangedDir = System.currentTimeMillis();
            }

            var elapsedTimeSinceLastDecredit = System.currentTimeMillis() - lastTimeDecreasedCredit;
            if (elapsedTimeSinceLastDecredit > 1000) {
                user.decreaseCredit(1);
                lastTimeDecreasedCredit = System.currentTimeMillis();
            }

            RideDTO rideDTO = new RideDTO(b.getId(), b.getLocation().x(), b.getLocation().y(), user.getCredit(), b.getBatteryLevel());
            eventBus.publish(rideUpdateAddress, rideDTO);

            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public void stopSimulation() {
        stopped = true;
        interrupt();
    }
}




this is my model 
package sap.ass01.layered.domain.model;

public class EBike {
    public enum EBikeState { AVAILABLE, IN_USE, MAINTENANCE }

    private final String id;
    private volatile EBikeState state;
    private volatile P2d location;
    private volatile V2d direction;
    private volatile double speed; // Units per simulation tick
    private volatile int batteryLevel; // 0..100

    public EBike(String id, double x, double y) {
        this.id = id;
        this.state = EBikeState.AVAILABLE;
        this.location = new P2d(x, y);
        this.direction = new V2d(1, 0); // Initial direction
        this.speed = 1.0; // Default speed
        this.batteryLevel = 100;
    }

    public String getId() { return id; }

    public synchronized EBikeState getState() { return state; }
    public synchronized void setState(EBikeState state) { this.state = state; }

    public synchronized P2d getLocation() { return location; }
    public synchronized void setLocation(P2d location) { this.location = location; }

    public synchronized V2d getDirection() { return direction; }
    public synchronized void setDirection(V2d direction) { this.direction = direction; }

    public synchronized double getSpeed() { return speed; }
    public synchronized void setSpeed(double speed) { this.speed = speed; }

    public synchronized int getBatteryLevel() { return batteryLevel; }
    public synchronized void decreaseBattery(int amount) {
        this.batteryLevel = Math.max(this.batteryLevel - amount, 0);
        if (this.batteryLevel == 0) {
            this.state = EBikeState.MAINTENANCE;
        }
    }

    public synchronized void rechargeBattery() {
        this.batteryLevel = 100;
        this.state = EBikeState.AVAILABLE;
    }

    public synchronized boolean isAvailable() {
        return this.state == EBikeState.AVAILABLE;
    }

    @Override
    public String toString() {
        return String.format("EBike{id='%s', location=%s, batteryLevel=%d%%, state='%s'}",
                id, location, batteryLevel, state);
    }
}

package sap.ass01.layered.domain.model;

import java.util.Date;
import java.util.Optional;

public class Ride {
    private final String id;
    private final User user;
    private final EBike ebike;
    private final Date startTime;
    private volatile Optional<Date> endTime;
    private volatile boolean ongoing;

    public Ride(String id, User user, EBike ebike) {
        this.id = id;
        this.user = user;
        this.ebike = ebike;
        this.startTime = new Date();
        this.endTime = Optional.empty();
        this.ongoing = false;
    }

    public String getId() { return id; }
    public User getUser() { return user; }
    public EBike getEbike() { return ebike; }
    public Date getStartTime() { return startTime; }
    public Optional<Date> getEndTime() { return endTime; }
    public boolean isOngoing() { return ongoing; }

    public void start() { this.ongoing = true; }

    public void end() {
        if (this.ongoing) {
            this.endTime = Optional.of(new Date());
            this.ongoing = false;
        }
    }

    @Override
    public String toString() {
        return String.format("Ride{id='%s', user='%s', ebike='%s', ongoing=%s}",
                id, user.getId(), ebike.getId(), ongoing);
    }
}

package sap.ass01.layered.domain.model;

public class P2d implements java.io.Serializable{

    private double x,y;

    public P2d(double x,double y){
        this.x=x;
        this.y=y;
    }

    public double x() {
        return x;
    }

    public double y() {
        return y;
    }

    public P2d sum(V2d v){
        return new P2d(x+v.x(),y+v.y());
    }

    public V2d sub(P2d v){
        return new V2d(x-v.x(),y-v.y());
    }

    public String toString(){
        return "P2d("+x+","+y+")";
    }
}

package sap.ass01.layered.domain.model;

import sap.ass01.layered.services.simulation.RideSimulation;

public record RideEntry(Ride ride, RideSimulation simulation) {
}
package sap.ass01.layered.domain.model;

public class User {
    public enum UserType { ADMIN, USER }

    private final String id;
    private final String name;
    private final UserType type;
    private volatile int credit;

    public User(String id, String name, UserType type) {
        this.id = id;
        this.name = name;
        this.type = type;
        this.credit = 100; // Default credit
    }

    public String getId() { return id; }
    public String getName() { return name; }
    public UserType getType() { return type; }

    public int getCredit() { return credit; }

    public void decreaseCredit(int amount) {
        this.credit = Math.max(this.credit - amount, 0);
    }

    public void increaseCredit(int amount) {
        this.credit += amount;
    }

    @Override
    public String toString() {
        return String.format("User{id='%s', name='%s', type='%s', credit=%d}", id, name, type, credit);
    }
}

package sap.ass01.layered.domain.model;

public class V2d implements java.io.Serializable {

    private double x,y;

    public V2d(double x,double y){
        this.x = x;
        this.y = y;
    }


    public double x() {
        return x;
    }

    public double y() {
        return y;
    }

    public V2d sum(V2d v){
        return new V2d(x+v.x,y+v.y);
    }

    public V2d rotate(double degree) {
        var rad = degree * Math.PI/180;
        var cs = Math.cos(rad);
        var sn = Math.sin(rad);
        var x1 = x * cs - y * sn;
        var y1 = x * sn + y * cs;
        var v = new V2d(x1, y1).getNormalized();
        return v;
    }

    public double abs(){
        return (double)Math.sqrt(x*x+y*y);
    }

    public V2d getNormalized(){
        double module=(double)Math.sqrt(x*x+y*y);
        return new V2d(x/module,y/module);
    }

    public V2d mul(double fact){
        return new V2d(x*fact,y*fact);
    }

    public String toString(){
        return "V2d("+x+","+y+")";
    }
}
package sap.ass01.layered.services.dto;

public record EBikeDTO(String id, double x, double y, int batteryLevel, String state) {
}
package sap.ass01.layered.services.dto;

public record RideDTO(String id, double x, double y, int credit, int charge) {
}
package sap.ass01.layered.services.dto;

public record UserDTO(String id, int credit, String permission) {
}

so u have to help me getting a simple solution in rxjava that can lead to the objective changin so the remaining classes serviceImpl that is the port for presentation layer to go into the business and the simulation to trigger update every tick but only for the specified user