danielkec

CREATE OR REPLACE PROCEDURE create_queue(queueName IN VARCHAR2, qType IN VARCHAR2) IS
BEGIN
    dbms_aqadm.create_queue_table('FRANK.'||queueName||'_TAB', qType);
    dbms_aqadm.create_queue('FRANK.'||queueName,'FRANK.'||queueName||'_TAB');
    dbms_aqadm.start_queue('FRANK.'||queueName);
END;
/

-- Setup example AQ queues FRANK.EXAMPLE_QUEUE_1, FRANK.EXAMPLE_QUEUE_2, FRANK.EXAMPLE_QUEUE_3
begin

danielkec

DECLARE
    enqueue_options    DBMS_AQ.ENQUEUE_OPTIONS_T;
    message_properties DBMS_AQ.MESSAGE_PROPERTIES_T;
    message_handle     RAW(16);
    msg                SYS.AQ$_JMS_TEXT_MESSAGE;
BEGIN
    msg := SYS.AQ$_JMS_TEXT_MESSAGE.construct;
    msg.set_text('HELLO PLSQL WORLD ! ' || TO_CHAR(sysdate, 'DD-MM-YY HH24:MI:SS'));
    DBMS_AQ.ENQUEUE(
            queue_name => 'FRANK.EXAMPLE_QUEUE_1',

danielkec

public static void main(String[] args) throws SQLException {
    System.setProperty("oracle.jdbc.fanEnabled", "false");

    PoolDataSource pds = PoolDataSourceFactory.getPoolDataSource();
    pds.setConnectionFactoryClassName("oracle.jdbc.pool.OracleDataSource");
    pds.setURL("jdbc:oracle:thin:@helidonaq_high?TNS_ADMIN=/home/kec/wallets/Wallet_helidonaq");
    pds.setUser("frank");
    pds.setPassword("SuperSecretPassword1234");

    AqConnector seConn = AqConnector.builder()

danielkec

    
    SubmissionPublisher<String> emitter = new SubmissionPublisher<>();
    SseBroadcaster sseBroadcaster;

    @Incoming("from-wls")
    public void receive(JmsMessage<String> msg) {
        if (sseBroadcaster == null) {
            System.out.println("No SSE client subscribed yet: " + msg.getPayload());
            return;
        }

danielkec

    incoming:
      from-wls-1:
        connector: helidon-jms
        destination: ./TestJMSModule!ms1@udd_queue
      from-wls-2:
        connector: helidon-jms
        destination: ./TestJMSModule!ms2@udd_queue
    outgoing:
      to-wls:
        connector: helidon-jms

danielkec

    
    @Incoming("from-wls-1")
    public void receive1(String payld) {
        sseBroadcaster.broadcast(new OutboundEvent.Builder().data("Q1: " + payld).build());
    }

    @Incoming("from-wls-2")
    public void receive2(String payld) {
        sseBroadcaster.broadcast(new OutboundEvent.Builder().data("Q2: " + payld).build());
    }

danielkec

 @Incoming("from-stream")
 public void receive(CloudEvent event) {
     String payload = new String(event.getData().toBytes());
 }

danielkec

final SubmissionPublisher<CloudEvent> emitter = new SubmissionPublisher<>();

@Outgoing("to-stream")
public PublisherBuilder<CloudEvent> registerEmitter() {
    return ReactiveStreams.fromPublisher(FlowAdapters.toPublisher(emitter));
}

@POST
@Path("/send/{msg}")
public void send(@PathParam("msg") String payload) {

danielkec

  connector:
    helidon-kafka:
      bootstrap.servers: ${oci.test-stream.endpoint}:${oci.test-stream.port}
      ...
      key.serializer: org.apache.kafka.common.serialization.StringSerializer
      key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
      
      value.serializer: io.cloudevents.kafka.CloudEventSerializer
      value.deserializer: io.cloudevents.kafka.CloudEventDeserializer

danielkec

        <dependency>
            <groupId>io.helidon.microprofile.messaging</groupId>
            <artifactId>helidon-microprofile-messaging</artifactId>
        </dependency>
        <dependency>
            <groupId>io.helidon.messaging.kafka</groupId>
            <artifactId>helidon-messaging-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>io.cloudevents</groupId>