Daemon écoutant une queue Oracle

Version imprimable

Bien le bonjour,

J'aimerais implémenter un daemon écoutant une queue Oracle, affichant chaque nouveau message reçu sur cette queue et enlevant le message de la queue après affichage.

Particularité : ma queue fait partie d'une queue table avec multiple consumers.

Je pensais faire tout cela à l'aide de l'API oracle.AQ mais il y a vraiment très peu de doc et encore moins d'exemples... Est-ce qu'il y a moyen de faire plus simple ou plus direct ?

Merci d'avance !

Le code SQL pour la création de la queue table TQ_AQ_INBOUND_FLOW :

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
BEGIN
  SYS.DBMS_AQADM.DROP_QUEUE_TABLE
    (QUEUE_TABLE          =>        'MILDJU.TQ_AQ_INBOUND_FLOW');
END;
/
 
BEGIN
  SYS.DBMS_AQADM.CREATE_QUEUE_TABLE
  (
    QUEUE_TABLE           =>        'MILDJU.TQ_AQ_INBOUND_FLOW'
   ,QUEUE_PAYLOAD_TYPE    =>        'SYS.AQ$_JMS_TEXT_MESSAGE'
   ,COMPATIBLE            =>        '10.0.0'
   ,STORAGE_CLAUSE        =>        '
                                     TABLESPACE TSPACE_001
                                     PCTUSED    0
                                     PCTFREE    10
                                     INITRANS   1
                                     MAXTRANS   255
                                     STORAGE    (
                                                 INITIAL          64K
                                                 NEXT             1M
                                                 MINEXTENTS       1
                                                 MAXEXTENTS       UNLIMITED
                                                 PCTINCREASE      0
                                                 BUFFER_POOL      DEFAULT
                                                )'
   ,SORT_LIST             =>        'ENQ_TIME'
   ,MULTIPLE_CONSUMERS    =>         TRUE
   ,MESSAGE_GROUPING      =>         0
   ,SECURE                =>         FALSE
   );
END;
/

Et celui de la queue à écouter :

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
BEGIN
  SYS.DBMS_AQADM.STOP_QUEUE ( QUEUE_NAME => 'MILDJU.AQ_INBOUND_FLOW');
  SYS.DBMS_AQADM.DROP_QUEUE ( QUEUE_NAME => 'MILDJU.AQ_INBOUND_FLOW');
END;
/
 
BEGIN
  SYS.DBMS_AQADM.CREATE_QUEUE
  (
    QUEUE_NAME          =>   'MILDJU.AQ_INBOUND_FLOW'
   ,QUEUE_TABLE         =>   'MILDJU.TQ_AQ_INBOUND_FLOW'
   ,QUEUE_TYPE          =>   SYS.DBMS_AQADM.NORMAL_QUEUE
   ,MAX_RETRIES         =>   5
   ,RETRY_DELAY         =>   0
   ,RETENTION_TIME      =>   0
   );
END;
/
 
BEGIN
  SYS.DBMS_AQADM.START_QUEUE
  (
    QUEUE_NAME => 'MILDJU.AQ_INBOUND_FLOW'
   ,ENQUEUE => TRUE 
   ,DEQUEUE => TRUE 
   );
END;
/

Toujours pas de réponse, mais j'ai bossé aussi un peu de mon côté... :mouarf:

J'ai trouvé quelques exemples sur le site d'Oracle mais à l'exécution, j'obtiens un code d'erreur JMS-174 (Class must be specified for queues with object payloads) et je ne sais pas pourquoi cela ne fonctionne pas. :cry:

Voici mon code :

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
public void toto() throws MildjuException {
    AQSession aq_sess = null;
    AQQueueTable q_table;
    AQQueue queue;
    AQMessage message;
    AQRawPayload raw_payload;
    AQDequeueOption deq_option;
    byte[] b_array;
    Connection db_conn  = null;
 
    try {
        aq_sess = createSession();
 
        db_conn = ((AQOracleSession)aq_sess).getDBConnection();
 
        // Get a handle to queue table - TQ_AQ_INBOUND_FLOW in MILDJU schema
        q_table = aq_sess.getQueueTable("MILDJU", "TQ_AQ_INBOUND_FLOW");
        System.out.println("Successful getQueueTable");  
 
        // Get a handle to a queue - AQ_INBOUND_FLOW in MILDJU schema
        queue = aq_sess.getQueue("MILDJU", "AQ_INBOUND_FLOW");
        System.out.println("Successful getQueue");  
 
        // Creating a AQDequeueOption object with default options
        deq_option = new AQDequeueOption();
 
        // Dequeue a message
        message = queue.dequeue(deq_option);
        System.out.println("Successful dequeue"); 
 
        // Retrieve raw data from the message
        raw_payload = message.getRawPayload();
 
        b_array = raw_payload.getBytes();
 
        System.out.println("Le message : " + b_array.toString());
 
        db_conn.commit();
    } catch (Exception e) {
        throw new MildjuException(e.getMessage());
    }
}
 
 
public static AQSession createSession() {
    Connection db_conn;
    AQSession  aq_sess = null;
 
    try {
        Class.forName("oracle.jdbc.driver.OracleDriver");
        db_conn = DriverManager.getConnection("jdbc:oracle:thin:@mamachine:1918:MA_BASE", "MILDJU", "MILDJU");
 
        System.out.println("JDBC Connection opened "); 
        db_conn.setAutoCommit(false);
 
        // Load the Oracle AQ driver
        Class.forName("oracle.AQ.AQOracleDriver");
 
        // Creating an AQ Session
        aq_sess = AQDriverManager.createAQSession(db_conn);
        System.out.println("Successfully created AQSession ");  
    } catch (Exception ex) {
        System.out.println("Exception: " + ex); 
        ex.printStackTrace();      
    }
 
    return aq_sess;
}

et le message d'erreur :

Citation:

JDBC Connection opened
Successfully created AQSession
Successful getQueueTable
Successful getQueue
=ERROR 24 Feb 13:16:07 - JMS-174: Class must be specified for queues with object payloads
Use dequeue(deq_option, payload_fact) or dequeue(deq_option, sql_data_cl)

28/02/2011, 13h56
Mildju

Vu que tout le monde semble s'y intéresser :mouarf: , je vais vous dire comment j'ai finalement pu y arriver. J'ai abandonné l'API oracle.AQ pour utiliser la javax.jms.

En gros, j'ai créé une TopicConnection à partir duquel j'ai créé une TopicSession, à partir duquel j'ai créé un Topic basé sur le nom de ma queue.

Ensuite, j'ai ajouté mon propre subscriber pour recevoir les messages. L'astuce, dans mon cas, était d'utiliser un durable subscriber permettant de récupérer tous les messages déjà présents sur la queue (et tous ceux qui doivent arriver par la suite).

Ne pas oublier le commit à la fin pour consommer le message.

Merci de votre attention, à vous les studios ! :P