ReactiveMessaging.java

package io.smallrye.reactive.messaging.example.eventclouds;

import io.smallrye.reactive.messaging.cloudevents.CloudEventMessage;
import io.smallrye.reactive.messaging.http.HttpMessage;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import javax.enterprise.context.ApplicationScoped;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

@ApplicationScoped
public class MyCloudEventProcessor {

// processor from "source" out to (external) http
  @Incoming("source")
  @Outgoing("to-http")
  public HttpMessage<CloudEventMessage>  process(CloudEventMessage<String> message) {
    return HttpMessage.HttpMessageBuilder.<CloudEventMessage>create()
      .withMethod("PUT")
      .withPayload(message)
      .withHeader("Content-Type", "application/cloudevents+json")
      .build();
  }

// my HTTP Server
  @Incoming("from-http")
  public CompletionStage<Void> receive(Message m) {
    System.out.println("da -> " + m);
    return CompletableFuture.completedFuture(null);
  }

}