INFO 1 --- [nio-8080-exec-1] i.l.cabs.common.cloudevents.Publisher    :
Publishing event to http://broker-ingress.knative-eventing.svc.cluster.local/demo/default :
CloudEvent{id='83720fe5-02ee-4a3e-9b22-5c287fb68d10',source=usvc://cabs/legacy,
type='cabs.drivers.calculate-fee', datacontenttype='application/json',
subject='4e630a96-4d5c-488c-a53b-9554c0bcb97e',time=2025-02-04T17:32:20.638351262Z,
data=BytesCloudEventData{value=[123, 34, 100, 114, 105, 118, 101, 114, 45, 105,
100, 34, 58, 49, 57, 57, 51, 52, 51, 50, 53, 53, 50, 44, 34, 116, 114, 97, 110,
115, 105, 116, 45, 112, 114, 105, 99, 101, 34, 58, 53, 49, 48, 48, 125]},
extensions={}}

[INFO  drivers::app::events] Received event:
    CloudEvent:
      specversion: '1.0'
      id: 'f94792bc-9c38-4db1-8da6-b6a28d1b4847'
      type: 'cabs.drivers.calculate-fee'
      source: 'usvc://cabs/legacy'
      datacontenttype: 'application/json'
      subject: '005be37e-8971-4a5b-b5e7-dd18de3c1184'
      time: '2025-02-04T17:48:11.641317948+00:00'
      knativearrivaltime: '2025-02-04T17:48:11.655926003Z'
      Binary data: "{\"driver-id\":1993432552,\"transit-price\":5100}"

[DEBUG drivers::drivers::service] calculate fee for: Subject {
 id: Some("005be37e-8971-4a5b-b5e7-dd18de3c1184"),
 entity: CalculateFeeEvent {
  driver_id: Identifier(1993432552),
  transit_price: Money(5100) } }
[DEBUG drivers::drivers::service] fee value: Money(4856)
[DEBUG drivers::support::cloudevents] sending cabs.drivers.driver-fee event to
 http://broker-ingress.knative-eventing.svc.cluster.local/demo/default:
  Event { attributes: V10(Attributes { id: "939babd7-6a85-4859-b45b-66087aba9418",
   ty: "cabs.drivers.driver-fee", source: "usvc://cabs/drivers",
   datacontenttype: Some("application/json"), dataschema: None,
   subject: Some("005be37e-8971-4a5b-b5e7-dd18de3c1184"),
   time: Some(2025-02-04T17:48:12.897943139Z) }),
   data: Some(Json(Object {"driver-id": Number(1993432552), "fee": Number(4856)})),
   extensions: {} }

INFO 1 --- [nio-8080-exec-2] i.l.c.ride.details.TransitDetailsFacade  :
 Driver fee calculated for transit 005be37e-8971-4a5b-b5e7-dd18de3c1184: 48.56

impl Service {
  pub async fn calculate_fee(&mut self, ce: Event) -> Result<()> {
    let calc_fee_intent = Self::unwrap_calculatefee(ce)?; (1)
    let subject = calc_fee_intent.id.clone();

    log::debug!("calculate fee for: {:?}", calc_fee_intent);
    let drv = self.repo.get(&calc_fee_intent.entity.driver_id).await?;

    let fee = drv.calculate_fee(&calc_fee_intent.entity.transit_price); (2)

    log::debug!("fee value: {:?}", fee);

    let driverfee_event = DriverFeeEvent {
        driver_id: calc_fee_intent.entity.driver_id,
        fee,
    }; (3)

    let mut builder = driverfee_event.to_builder(); (3)
    if let Some(id) = subject {
        builder = builder.subject(id);
    } (3)
    let ce = builder.build().map_err(error::ErrorInternalServerError)?; (3)

    Sender::new(&self.config).send(ce).await?; (4)

    Ok(())
  }
  // [..]
}

pub fn routes() -> impl HttpServiceFactory + 'static {
    web::resource("/").route(web::post().to(recv))
}

async fn recv(
    ce: Event,
    state: web::Data<State>,
    binding: web::Data<Binding>,
) -> Result<HttpResponse> {
    log::info!("Received event:\n{}", ce);

    let mut svc = service::new(state, binding).await?;

    match ce.ty() {
        "cabs.drivers.calculate-fee" => svc.calculate_fee(ce).await,
        _ => Err(error::ErrorBadRequest("unsupported event type")),
    }?;

    Ok(HttpResponse::Ok().finish())
}

impl Sender {
    pub async fn send(&self, ce: Event) -> Result<()> {
        log::debug!("sending {} event to {}:\n{:?}", ce.ty(), &self.sink, ce,);

        let response = self
            .client
            .post(&self.sink) (1)
            .event(ce)
            .map_err(error::ErrorInternalServerError)?
            .send()
            .await
            .map_err(error::ErrorInternalServerError)?;

        match response.status().is_success() {
            true => Ok(()),
            false => {
                log::error!("failed to send event: {:#?}", response);
                Err(error::ErrorInternalServerError(format!(
                    "failed to send event: {}",
                    response.status()
                )))
            }
        }
    }
}

apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
  name: default
  namespace: demo
spec:
  delivery:
    backoffDelay: PT0.2S (1)
    backoffPolicy: exponential (2)
    retry: 10 (3)

public void completeTransit(UUID requestUUID, AddressDTO destinationAddress) {
    // ...
    Money finalPrice = completeTransitService.completeTransit(driverId, requestUUID, from, to);
    // ...
    driverFeeService.calculateDriverFee(requestUUID, finalPrice, driverId); (1)
    // ...
}

@EventListener (2)
public void driverFeeCalculated(DriverFee driverFee) { (3)
    Objects.requireNonNull(driverFee.ctx.getSubject());
    UUID id = UUID.fromString(driverFee.ctx.getSubject());
    transitDetailsFacade.driverFeeCalculated(id, driverFee.data.fee);
}

@Service
public class DriverFeeService {

    private final CloudEventSender eventSender;

    @Autowired
    public DriverFeeService(EventSender eventSender) {
        this.eventSender = eventSender;
    }

    public void calculateDriverFee(UUID rideId, Money transitPrice, Long driverId) {
        eventSender.send(new CalculateFee(
            rideId,
            driverId,
            transitPrice.toInt()
        ));
    }
}

@Service
public class CloudEventSender {

  private static final Logger log = LoggerFactory.getLogger(EventSender.class);

  private final KnativeConfig knative;
  private final List<Into<?>> converters;

  @Autowired
  CloudEventSender(KnativeConfig knative, List<Into<?>> converters) {
    this.knative = knative;
    this.converters = converters;
  }

  public void send(Object event) {
    try {
      unsafeSend(event);
    } catch (IOException e) {
      throw new UncheckedIOException(e);
    }
  }

  private <T> void unsafeSend(T event) throws IOException {
    Into<T> convert = (Into<T>) converters.stream()
        .filter(c -> c.accepts(event))
        .findFirst()
        .orElseThrow(() -> new IllegalArgumentException(
          "Cannot find converter for " + event.getClass()));
    CloudEvent ce = convert.into(event);
    URL url = knative.getSink();
    log.info("Publishing event to {} : {}", url, ce);
    HttpURLConnection http = (HttpURLConnection) url.openConnection();
    http.setRequestMethod("POST");
    http.setDoOutput(true);
    http.setDoInput(true);

    HttpMessageWriter messageWriter = createMessageWriter(http);
    messageWriter.writeBinary(ce);

    int code = http.getResponseCode();
    if (code < 200 || code >= 300) {
      throw new IOException("Unexpected response code " + code);
    }
  }
}

@RestController
public class CloudEventReceiver {
  private static final Logger log = LoggerFactory.getLogger(Receiver.class);

  private final EventsPublisher eventsPublisher;
  private final List<From<?>> froms;

  @Autowired
  Receiver(EventsPublisher eventsPublisher, List<From<?>> froms) {
    this.eventsPublisher = eventsPublisher;
    this.froms = froms;
  }

  @PostMapping("/")
  public void receive(@RequestBody CloudEvent event) {
    log.info("Received event: {}", event);

    for (From<?> from : froms) {
      if (from.matches(event)) {
        Event ev = from.fromCloudEvent(event); (1)
        eventsPublisher.publish(ev); (2)
        return;
      }
    }

    throw new IllegalStateException("No matching event type consumer found");
  }
}

apiVersion: sources.knative.dev/v1
kind: SinkBinding
metadata:
  name: drivers-binding
  namespace: demo
spec:
  sink:
    ref:
      apiVersion: eventing.knative.dev/v1
      kind: Broker
      name: default
      namespace: demo
  subject:
    apiVersion: serving.knative.dev/v1
    kind: Service
    name: drivers
    namespace: demo
---
apiVersion: sources.knative.dev/v1
kind: SinkBinding
metadata:
  name: legacy-binding
  namespace: demo
spec:
  sink:
    ref:
      apiVersion: eventing.knative.dev/v1
      kind: Broker
      name: default
      namespace: demo
  subject:
    apiVersion: serving.knative.dev/v1
    kind: Service
    name: legacy
    namespace: demo

apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: trg-drivers
  namespace: demo
spec:
  broker: default
  filter:
    attributes:
      type: cabs.drivers.calculate-fee (1)
  subscriber:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: drivers
      namespace: demo
---
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: trg-legacy
  namespace: demo
spec:
  broker: default
  filter:
    attributes:
      type: cabs.drivers.driver-fee (1)
  subscriber:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: legacy
      namespace: demo