【PL/SQL】DBMS_ALERTとDBMS_PIPEの使い方｜セッション間通知・非同期メッセージ・実務注意点

DECLARE
  v_message VARCHAR2(1800);
  v_status  PLS_INTEGER;
BEGIN
  DBMS_ALERT.REGISTER('ORDER_CHANGED');

  DBMS_ALERT.WAITONE(
    name    => 'ORDER_CHANGED',
    message => v_message,
    status  => v_status,
    timeout => 30
  );

  IF v_status = 0 THEN
    DBMS_OUTPUT.PUT_LINE('message=' || v_message);
  ELSE
    DBMS_OUTPUT.PUT_LINE('timeout or no alert');
  END IF;

  DBMS_ALERT.REMOVE('ORDER_CHANGED');
END;
/

BEGIN
  UPDATE orders
  SET status = 'READY'
  WHERE order_id = 1001;

  DBMS_ALERT.SIGNAL(
    name    => 'ORDER_CHANGED',
    message => 'order_id=1001'
  );

  COMMIT; -- この時点で待機側に通知される
END;
/

-- 重要なイベントはテーブルに保存する
INSERT INTO app_event_queue (
  event_id,
  event_type,
  payload,
  status,
  created_at
) VALUES (
  app_event_seq.NEXTVAL,
  'ORDER_CHANGED',
  'order_id=1001',
  'READY',
  SYSTIMESTAMP
);

DBMS_ALERT.SIGNAL('APP_EVENT_READY', 'ORDER_CHANGED');

COMMIT;

DECLARE
  v_message VARCHAR2(1800);
  v_status  PLS_INTEGER;
BEGIN
  DBMS_ALERT.REGISTER('APP_EVENT_READY');

  LOOP
    DBMS_ALERT.WAITONE('APP_EVENT_READY', v_message, v_status, 60);

    IF v_status = 0 THEN
      FOR r IN (
        SELECT event_id, event_type, payload
        FROM app_event_queue
        WHERE status = 'READY'
        ORDER BY event_id
      ) LOOP
        UPDATE app_event_queue
        SET status = 'DONE',
            processed_at = SYSTIMESTAMP
        WHERE event_id = r.event_id
          AND status = 'READY';
      END LOOP;

      COMMIT;
    ELSE
      DBMS_OUTPUT.PUT_LINE('wait timeout');
    END IF;
  END LOOP;
END;
/

DECLARE
  v_status PLS_INTEGER;
BEGIN
  DBMS_PIPE.RESET_BUFFER;
  DBMS_PIPE.PACK_MESSAGE('REFRESH_CACHE');
  DBMS_PIPE.PACK_MESSAGE('product');
  DBMS_PIPE.PACK_MESSAGE(1001);

  v_status := DBMS_PIPE.SEND_MESSAGE(
    pipename => 'APP_WORKER_PIPE',
    timeout  => 10
  );

  IF v_status <> 0 THEN
    RAISE_APPLICATION_ERROR(-20001, 'pipe send failed: ' || v_status);
  END IF;
END;
/

DECLARE
  v_status  PLS_INTEGER;
  v_command VARCHAR2(100);
  v_target  VARCHAR2(100);
  v_id      NUMBER;
BEGIN
  v_status := DBMS_PIPE.RECEIVE_MESSAGE(
    pipename => 'APP_WORKER_PIPE',
    timeout  => 30
  );

  IF v_status = 0 THEN
    DBMS_PIPE.UNPACK_MESSAGE(v_command);
    DBMS_PIPE.UNPACK_MESSAGE(v_target);
    DBMS_PIPE.UNPACK_MESSAGE(v_id);

    DBMS_OUTPUT.PUT_LINE(v_command || ':' || v_target || ':' || v_id);
  ELSE
    DBMS_OUTPUT.PUT_LINE('timeout or receive error: ' || v_status);
  END IF;
END;
/

DECLARE
  v_status PLS_INTEGER;
BEGIN
  v_status := DBMS_PIPE.CREATE_PIPE(
    pipename    => 'APP1.PROD.WORKER_CMD',
    maxpipesize => 8192,
    private     => TRUE
  );

  IF v_status <> 0 THEN
    RAISE_APPLICATION_ERROR(-20002, 'create pipe failed: ' || v_status);
  END IF;
END;
/

DECLARE
  v_status PLS_INTEGER;
BEGIN
  v_status := DBMS_PIPE.REMOVE_PIPE('APP1.PROD.WORKER_CMD');
  DBMS_OUTPUT.PUT_LINE('remove status=' || v_status);
END;
/

DECLARE
  v_message VARCHAR2(1800);
  v_status  PLS_INTEGER;
  v_stop    NUMBER;
BEGIN
  DBMS_ALERT.REGISTER('APP_EVENT_READY');

  LOOP
    SELECT COUNT(*)
    INTO v_stop
    FROM app_control
    WHERE control_name = 'WORKER_STOP'
      AND control_value = 'Y';

    EXIT WHEN v_stop > 0;

    DBMS_ALERT.WAITONE('APP_EVENT_READY', v_message, v_status, 30);

    IF v_status = 0 THEN
      DBMS_OUTPUT.PUT_LINE('received: ' || v_message);
    END IF;
  END LOOP;

  DBMS_ALERT.REMOVE('APP_EVENT_READY');
END;
/

CREATE OR REPLACE TRIGGER orders_notify_trg
AFTER INSERT OR UPDATE ON orders
BEGIN
  DBMS_ALERT.SIGNAL(
    name    => 'ORDER_CHANGED',
    message => 'orders changed'
  );
END;
/

CREATE TABLE app_ipc_log (
  log_id      NUMBER GENERATED BY DEFAULT AS IDENTITY,
  channel     VARCHAR2(100),
  direction   VARCHAR2(10),
  status_code NUMBER,
  message     VARCHAR2(1800),
  log_time    TIMESTAMP DEFAULT SYSTIMESTAMP,
  CONSTRAINT pk_app_ipc_log PRIMARY KEY (log_id)
);

INSERT INTO app_ipc_log(channel, direction, status_code, message)
VALUES ('APP_EVENT_READY', 'RECEIVE', v_status, v_message);