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

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

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

create user frank identified by SuperSecretPassword1234;

grant connect to frank;
grant resource to frank;
grant execute on dbms_aq to frank;
grant execute on dbms_aqadm to frank;
grant execute on dbms_aqin to frank;
grant unlimited tablespace to frank;

BEGIN

danielkec

    private AtomicInteger counter = new AtomicInteger();

    @Incoming("from-aq")
    @Acknowledgment(Acknowledgment.Strategy.MANUAL)
    public CompletionStage<?> fromAq(AqMessage<String> msg) {
        System.out.println("Received: " + msg.getPayload());
        if (counter.getAndIncrement() == 5) {
            throw new RuntimeException("5th message exception!");
        }
        //Acknowledgement/commit is called after the business code

danielkec

    @Outgoing("to-aq")
    public Publisher<String> toAq() {
        return FlowAdapters.toPublisher(
                Multi.interval(2, TimeUnit.SECONDS, Executors.newSingleThreadScheduledExecutor())
                        .map(i -> "Message " + i)
        );
    }


    @Incoming("from-aq")

danielkec

 <dependency>
     <groupId>io.helidon.microprofile.messaging</groupId>
     <artifactId>helidon-microprofile-messaging</artifactId>
 </dependency>
 <dependency>
     <groupId>io.helidon.messaging.aq</groupId>
     <artifactId>helidon-messaging-aq</artifactId>
 </dependency>
 <dependency>
     <groupId>io.helidon.integrations.cdi</groupId>

danielkec

javax:
  sql:
    DataSource:
      local-example-ds:
        connectionFactoryClassName: oracle.jdbc.pool.OracleDataSource
        URL: jdbc:oracle:thin:@localhost:1521:XE
        user: frank
        password: frank

mp:

danielkec

 // Setup connector
 JmsConnector seConn = JmsConnector.create();

 // Prepare channels
 Channel<String> toJms = Channel.create("to-jms");
 Channel<String> fromJms = Channel.create("from-jms");

 // Prepare emitter for interaction with non-reactive code
 Emitter<String> emitter = Emitter.create(toJms);

danielkec

 <dependency>
     <groupId>io.helidon.messaging.jms</groupId>
     <artifactId>helidon-messaging-jms</artifactId>
 </dependency>
 <!-- JMS client implenting JMS api of your choice-->
 <dependency>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-client</artifactId>
 </dependency>