OCI Queue Post Messages to the Queue

oci_queue_consume_msg_example.sql

DECLARE

  lt_parm_names     apex_application_global.VC_ARR2;
  lt_parm_values    apex_application_global.VC_ARR2;
  l_response        CLOB;
  l_rest_api_url    VARCHAR2(255);
  l_item_obj        json_object_t := json_object_t();
  l_json_obj        json_object_t := json_object_t();
  l_msgs_arr        json_array_t  := json_array_t();
  l_del_json_obj    json_object_t := json_object_t();
  l_del_entries_arr json_array_t  := json_array_t();
  l_del_receipt_obj json_object_t := json_object_t();
  l_del_payload     CLOB;
  l_msg_count       PLS_INTEGER;
  l_receipt         VARCHAR2(255);
  lt_receipts       apex_t_varchar2;
  lr_items          demo_master_items%ROWTYPE;

BEGIN

  -- Build the URL for the POST messages API.
  -- Change the <OCID> for your queue OCID.
  l_rest_api_url := 'https://cell-1.queue.messaging.us-ashburn-1.oci.oraclecloud.com/20210201/queues/<OCID>/messages';

  -- Set the HTTP Headers for the REST API Call.
  apex_web_service.g_request_headers.DELETE;
  apex_web_service.g_request_headers(1).NAME := 'Content-Type';
  apex_web_service.g_request_headers(1).VALUE := 'application/json';

  -- Set Parameter to fetch 20 messages at a time.
  lt_parm_names.DELETE;
  lt_parm_names(1)  := 'limit';
  lt_parm_values(1) := '20';

  -- Call Messages REST API to get visible messages from the Queue.
  l_response := apex_web_service.make_rest_request
    (p_url                  => l_rest_api_url,
     p_parm_name            => lt_parm_names,
     p_parm_value           => lt_parm_values,
     p_http_method          => 'GET',     
     p_credential_static_id => 'APEX_OCI_BLOG_CREDENTIAL');
  apex_automation.log_info (p_message => 'Get Messages Status ['||apex_web_service.g_status_code||'], Response ['||l_response||']');

  IF apex_web_service.g_status_code <> 200 THEN
    -- Raise an because the REST API Call Failed.
    apex_automation.log_error (p_message => 'Unable to Get Messages. Status ['||apex_web_service.g_status_code||
                              '], Response: ' || l_response);
    raise_application_error (-20000, 'Unable to Delete Queue Items.');
  END IF;

  -- Parse the JSON Response.
  l_json_obj := json_object_t.parse(l_response);
  
  -- Get the messages array
  l_msgs_arr := l_json_obj.get_Array('messages');
  
  -- Get a count of messages
  l_msg_count := l_msgs_arr.get_size;
  apex_automation.log_info (p_message => 'Found ['||l_msg_count||'] Messages on Queue');
  
  IF l_msg_count > 0 THEN
    -- Loop through messages.
    FOR i IN 0..l_msg_count -1 LOOP
      -- Get the receipt for the message.
      l_receipt := json_object_t(l_msgs_arr.get(i)).get_string('receipt');
      apex_automation.log_info (p_message => 'Processing Receipt ['||l_receipt||'], JSON ['||json_object_t(l_msgs_arr.get(i)).get_clob('content')||']');
      -- Parse the JSON in the content object.
      -- This is the payload that was added to the queue initially.
      l_item_obj := json_object_t.parse(json_object_t(l_msgs_arr.get(i)).get_clob('content'));
      -- Populate the Item Record from the Parsed JSON.
      lr_items.item_number      := l_item_obj.get_String('item_number');
      lr_items.item_description := l_item_obj.get_string('item_description');
      lr_items.country_of_mfg   := l_item_obj.get_string('country_of_mfg');
      lr_items.category_code    := l_item_obj.get_string('category_code');

      -- Create Record in Master Instance table.
      INSERT INTO DEMO_MASTER_ITEMS VALUES lr_items;

      -- Add Receipt to Array for Later Deletion from the Queue. 
      apex_string.push(lt_receipts, l_receipt);
    
    END LOOP;
  
    -- Build a JSON Array of receipts to pass to the deleteMessages REST API.
    FOR i IN 1..l_msg_count LOOP
      l_del_receipt_obj.put('receipt', lt_receipts(i));
      l_del_entries_arr.append(l_del_receipt_obj);
    END LOOP;
    l_del_json_obj.put('entries', l_del_entries_arr);
    l_del_payload := l_del_json_obj.to_clob();
    apex_automation.log_info (p_message => 'Delete Messages Payload ['||l_del_payload||']');
    apex_debug.info('XX Delete Messages Payload [%s]', l_del_payload);

    -- Set the HTTP Headers for the deleteMessages REST API Call.
    apex_web_service.g_request_headers.DELETE;
    apex_web_service.g_request_headers(1).NAME := 'Content-Type';
    apex_web_service.g_request_headers(1).VALUE := 'application/json';
  
    -- Call the deleteMessages REST API to delete Consumed Messages from the Queue
    -- Change the <OCID> for your queue OCID.
    l_rest_api_url := 'https://cell-1.queue.messaging.us-ashburn-1.oci.oraclecloud.com/20210201/queues/<OCID>/messages/actions/deleteMessages';
    l_response := apex_web_service.make_rest_request
      (p_url                  => l_rest_api_url,
       p_http_method          => 'POST',
       p_body                 => l_del_payload,
       p_credential_static_id => 'APEX_OCI_BLOG_CREDENTIAL');
    apex_automation.log_info (p_message => 'Delete Messages Status ['||apex_web_service.g_status_code||'], Response ['||l_response||']');

    IF apex_web_service.g_status_code <> 200 THEN
      -- Raise an Error as the Item Create/Update was not added too the Queue.
      apex_automation.log_error (p_message => 'Unable to Delete Queue Items. Status ['||apex_web_service.g_status_code||
                                '], Response: ' || l_response);
      raise_application_error (-20001, 'Unable to Delete Queue Items.');
    END IF;
  END IF;
END;