It is a good practice to encapsulate business logic inside Service methods so that controllers and repositories are loosely coupled. The following is a Service written for encapsulating business logic for Taxi, and is available in spring-boot-2-taxi-service:

@Service
public class TaxiService {

 private final ReactiveRedisTemplate<String, String> reactiveRedisTemplate;
 private final TaxiRepository taxiRepository;
 private final LocationToPointConverter locationToPointConverter = new 
 LocationToPointConverter();

 public TaxiService(ReactiveRedisTemplate<String, String> 
 reactiveRedisTemplate, TaxiRepository taxiRepository) {
 this.reactiveRedisTemplate = reactiveRedisTemplate;
 this.taxiRepository = taxiRepository;
 }

 public Mono<Taxi> register(TaxiRegisterEventDTO taxiRegisterEventDTO) {
 Taxi taxi = new Taxi(taxiRegisterEventDTO.getTaxiId(), 
 taxiRegisterEventDTO.getTaxiType(), TaxiStatus.AVAILABLE);
 return Mono.just(taxiRepository.save(taxi));
 }

The preceding register method saves Taxi in the system so that it can fulfill rides. This will return a result that is a reactive single object:

public Mono<Taxi> updateLocation(String taxiId, LocationDTO locationDTO) {
 Optional<Taxi> taxiOptional = taxiRepository.findById(taxiId);
 if (taxiOptional.isPresent()) {
 Taxi taxi = taxiOptional.get();
 return reactiveRedisTemplate.opsForGeo().add(taxi.getTaxiType().toString(), locationToPointConverter.convert(locationDTO), taxiId.toString()).flatMap(l -> Mono.just(taxi));
 } else {
 throw getTaxiIdNotFoundException(taxiId);
 }
 }

The preceding updateLocation method will use the ReactiveRedisTemplate.opsForGeo().add method to update the location and taxiId of a Taxi grouped into taxi type. This will return a result that is a reactive single object:

    public Flux<GeoResult<RedisGeoCommands.GeoLocation<String>>> 
     getAvailableTaxis(TaxiType taxiType, Double latitude, Double longitude, 
     Double radius) {
        return reactiveRedisTemplate.opsForGeo().radius(taxiType.toString(), new 
         Circle(new Point(longitude, latitude), new Distance(radius, 
          Metrics.KILOMETERS)));
    }

The preceding getAvailableTaxis method will return all the Taxi IDs falling inside of a circle which has a center geo coordinate depicted by latitude, longitude, and radius in kilometers. This will return a result that is a reactive collection of objects:

    public Mono<TaxiStatus> getTaxiStatus(String taxiId) {
        Optional<Taxi> taxiOptional = taxiRepository.findById(taxiId);
        if (taxiOptional.isPresent()) {
            Taxi taxi = taxiOptional.get();
            return Mono.just(taxi.getTaxiStatus());
        } else {
            throw getTaxiIdNotFoundException(taxiId);
        }

    }

The preceding getTaxiStatus method will return TaxiStatus of a taxi identified by taxiId. This will return a result that is a reactive single object:

public Mono<Taxi> updateTaxiStatus(String taxiId, TaxiStatus taxiStatus) {
        Optional<Taxi> taxiOptional = taxiRepository.findById(taxiId);
        if (taxiOptional.isPresent()) {
            Taxi taxi = taxiOptional.get();
            taxi.setTaxiStatus(taxiStatus);
            return Mono.just(taxiRepository.save(taxi));
        } else {
            throw getTaxiIdNotFoundException(taxiId);
        }
    }

    private TaxiIdNotFoundException getTaxiIdNotFoundException(String taxiId) {
        return new TaxiIdNotFoundException("Taxi Id "+taxiId+" Not Found");
    }
} 

The preceding updateTaxiStatus method will update the TaxiStatus of a taxi identified by the taxiId. This will return a result that is a reactive single object.

The TaxiService class in the preceding code is annotated with the @Service stereotype annotation to mark it as a Spring Service. All methods of this service return either a Mono or Flux, enabling those to be used Reactively. When a Taxi specified by an ID does not exist, TaxiIdNotFoundException is thrown. The following is the implementation for it:

public class TaxiIdNotFoundException extends RuntimeException {
    public TaxiIdNotFoundException(String message) {
        super(message);
    }

    public TaxiIdNotFoundException(String message, Throwable cause) {
        super(message, cause);
    }
}

TaxiIdNotFoundException extends from the Exception class. Likewise, the following TaxiBookingService class is used for TaxiBooking, which is available in spring-boot-2-taxi-book-service:

@Service
public class TaxiBookingService {

    private final static Logger LOGGER = 
    LoggerFactory.getLogger(TaxiBookingService.class);

    private final RedisTemplate<String, String> redisTemplate;
    private final ReactiveRedisTemplate<String, String> reactiveRedisTemplate;
    private final TaxiBookingRepository taxiBookingRepository;
    private final ObjectMapper objectMapper = new ObjectMapper();
    private final LocationToPointConverter locationToPointConverter = new 
    LocationToPointConverter();

    public TaxiBookingService(RedisTemplate<String, String> redisTemplate, 
    ReactiveRedisTemplate<String, String> reactiveRedisTemplate, 
    TaxiBookingRepository taxiBookingRepository) {
        this.redisTemplate = redisTemplate;
        this.reactiveRedisTemplate = reactiveRedisTemplate;
        this.taxiBookingRepository = taxiBookingRepository;
    }

    public Mono<TaxiBooking> book(TaxiBookedEventDTO taxiBookedEventDTO) {
        TaxiBooking taxiBooking = new TaxiBooking();
        taxiBooking.setEnd(locationToPointConverter.convert(taxiBookedEventDTO.getEnd()));
        taxiBooking.setStart(locationToPointConverter.convert(taxiBookedEventDTO.getStart()));
        taxiBooking.setBookedTime(taxiBookedEventDTO.getBookedTime());
        taxiBooking.setCustomerId(taxiBookedEventDTO.getCustomerId());
        taxiBooking.setBookingStatus(TaxiBookingStatus.ACTIVE);
        TaxiBooking savedTaxiBooking = taxiBookingRepository.save(taxiBooking);
        return reactiveRedisTemplate.opsForGeo().add(getTaxiTypeBookings(taxiBookedEventDTO.getTaxiType()), taxiBooking.getStart(), taxiBooking.getTaxiBookingId()).flatMap(l -> Mono.just(savedTaxiBooking));
    }
   

The preceding book method will enable a passenger to save a TaxiBooking in Redis based on the supplied TaxiBookedEventDTO, and will use the reactiveRedisTemplate.opsForGeo().add() method to add the taxi booking by its type, to be listed by its starting location and taxiBookingId so that it can be queried using geo-location search queries:

public Mono<TaxiBooking> cancel(String taxiBookingId, TaxiBookingCanceledEventDTO canceledEventDTO) {
        Optional<TaxiBooking> taxiBookingOptional = 
        taxiBookingRepository.findById(taxiBookingId);
        if (taxiBookingOptional.isPresent()) {
            TaxiBooking taxiBooking = taxiBookingOptional.get();
            taxiBooking.setBookingStatus(TaxiBookingStatus.CANCELLED);
            taxiBooking.setReasonToCancel(canceledEventDTO.getReason());
            taxiBooking.setCancelTime(canceledEventDTO.getCancelTime());
            return Mono.just(taxiBookingRepository.save(taxiBooking));
        } else {
            throw getTaxiBookingIdNotFoundException(taxiBookingId);
        }
}    

The preceding cancel method retrieves a Taxi Booking by its taxiBookingId and updates the booking status to canceled along with the reason and canceled time:

public Mono<TaxiBooking> accept(String taxiBookingId, TaxiBookingAcceptedEventDTO acceptedEventDTO) {
        Optional<TaxiBooking> taxiBookingOptional = 
        taxiBookingRepository.findById(taxiBookingId);
        if (taxiBookingOptional.isPresent()) {
            TaxiBooking taxiBooking = taxiBookingOptional.get();
            taxiBooking.setTaxiId(acceptedEventDTO.getTaxiId());
            taxiBooking.setAcceptedTime(acceptedEventDTO.getAcceptedTime());
            return 
            Mono.just(taxiBookingRepository.save(taxiBooking)).doOnSuccess(t -> 
           {
                try {
                    redisTemplate.convertAndSend(RedisConfig.ACCEPTED_EVENT_CHANNEL, objectMapper.writeValueAsString(acceptedEventDTO));
                } catch (JsonProcessingException e) {
                    LOGGER.error("Error while sending message to Channel {}", 
                    RedisConfig.ACCEPTED_EVENT_CHANNEL, e);
                }
            });

        } else {
            throw getTaxiBookingIdNotFoundException(taxiBookingId);
        }
    }

The preceding accept method will enable a driver to accept a TaxiBooking to fulfill the ride. After updating the taxiId and acceptedTime the taxiBooking will be saved and a Booking Accepted Event will be triggered to notify any listeners:

    public Flux<GeoResult<RedisGeoCommands.GeoLocation<String>>> 
     getBookings(TaxiType taxiType, Double latitude, Double longitude, Double 
     radius) {
        return 
         reactiveRedisTemplate.opsForGeo().radius(getTaxiTypeBookings(taxiType), 
         new Circle(new Point(longitude, latitude), new Distance(radius, 
         Metrics.KILOMETERS)));
    }    

The preceding getBookings method will return TaxiBooking by taxiType, geo-location, and radius. So any taxiBookings that match the type and fall inside the circle whose center is marked by the geo-coordinates and has the same radius:

public Mono<TaxiBooking> updateBookingStatus(String taxiBookingId, TaxiBookingStatus taxiBookingStatus) {
        Optional<TaxiBooking> taxiBookingOptional = 
         taxiBookingRepository.findById(taxiBookingId);
        if (taxiBookingOptional.isPresent()) {
            TaxiBooking taxiBooking = taxiBookingOptional.get();
            taxiBooking.setBookingStatus(taxiBookingStatus);
            return Mono.just(taxiBookingRepository.save(taxiBooking));
        } else {
            throw getTaxiBookingIdNotFoundException(taxiBookingId);
        }
    }

The preceding updateBookingStatus will update the bookingStatus of a TaxiBooking identified by the taxiBookingId passed in:

    private TaxiBookingIdNotFoundException 
     getTaxiBookingIdNotFoundException(String taxiBookingId) {
        return new TaxiBookingIdNotFoundException("Taxi Booking Id 
        "+taxiBookingId+" Not Found");
    }

    private String getTaxiTypeBookings(TaxiType taxiType) {
        return taxiType.toString()+"-Bookings";
    }

}

The TaxiBookingService in the preceding code is also annotated with @Service stereotype annotation to mark it as a Spring Service. All methods of this service return either a Mono or Flux, enabling those to be used Reactively. When a Taxi booking specified by an ID does not exist, TaxiBookingIdNotFoundException is thrown. Following is the implementation:

public class TaxiBookingIdNotFoundException extends RuntimeException {

    public TaxiBookingIdNotFoundException(String message) {
        super(message);
    }

    public TaxiBookingIdNotFoundException(String message, Throwable cause) {
        super(message, cause);
    }

}

TaxiBookingIdNotFoundException extends from Exception .