Use the DBMS_PIPE package to transfer messages - PolarDB

DBMS_PIPE is a built-in package that lets sessions connected to the same cluster communicate by passing messages through named pipes.

Usage notes

Warning

Do not use DBMS_PIPE with cluster endpoints. Doing so may cause long-running transactions.

Subprograms

Subprogram	Description
CREATE_PIPE function	Creates a named pipe
NEXT_ITEM_TYPE function	Returns the data type code of the next item in the local message buffer
PACK_MESSAGE procedures	Adds a data item to the local message buffer
UNPACK_MESSAGE procedures	Retrieves the next data item from the local message buffer
PURGE procedure	Discards all messages in a named pipe
RECEIVE_MESSAGE function	Copies a message from a named pipe into the local message buffer
REMOVE_PIPE function	Removes a named pipe
RESET_BUFFER procedure	Clears the local message buffer
SEND_MESSAGE function	Sends the contents of the local message buffer to a named pipe
UNIQUE_SESSION_NAME function	Returns a name that is unique to the current session

CREATE_PIPE function

Creates a named pipe. Use this function to create an explicit pipe before sending or receiving messages.

Syntax

DBMS_PIPE.CREATE_PIPE(
   pipename     IN VARCHAR2,
   maxpipesize  IN INTEGER DEFAULT 8192,
   private      IN BOOLEAN DEFAULT TRUE
) RETURN INTEGER;

Parameters

Parameter	Description
`pipename`	Name of the pipe to create
`maxpipesize`	(Optional) Maximum capacity of the pipe, in bytes. Default: `8192`
`private`	(Optional) `TRUE` creates a private pipe (default). `FALSE` creates a public pipe

Return values

Return value	Description
`0`	Pipe created successfully

Examples

Create a private pipe (default)

DECLARE
    status INTEGER;
BEGIN
    status := DBMS_PIPE.CREATE_PIPE('messages');
    DBMS_OUTPUT.PUT_LINE('CREATE_PIPE status: ' || status);
END;

Expected output:

CREATE_PIPE status: 0

Create a public pipe

DECLARE
    status INTEGER;
BEGIN
    status := DBMS_PIPE.CREATE_PIPE('mailbox', 8192, FALSE);
    DBMS_OUTPUT.PUT_LINE('CREATE_PIPE status: ' || status);
END;

Expected output:

CREATE_PIPE status: 0

NEXT_ITEM_TYPE function

Returns the data type code of the next item in the local message buffer. Use this after calling RECEIVE_MESSAGE to determine the type of each item before unpacking it.

Syntax

DBMS_PIPE.NEXT_ITEM_TYPE() RETURN INTEGER;

Return values

Return value	Description
INTEGER	The data type of the next data item in the buffer

Example

This example uses two sessions. Session A packs and sends a VARCHAR2 message; Session B receives it and checks the item type before unpacking.

Session A

DECLARE
    str    VARCHAR2 := 'Character data';
    status INTEGER;
BEGIN
    DBMS_PIPE.PACK_MESSAGE(str);

    status := DBMS_PIPE.SEND_MESSAGE('messages');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status is: ' || status);
    -- SEND_MESSAGE status is: 0
END;

Session B

DECLARE
    str    VARCHAR2;
    status INTEGER;
BEGIN
    status := DBMS_PIPE.RECEIVE_MESSAGE('messages');
    DBMS_OUTPUT.PUT_LINE('RECEIVE_MESSAGE status is: ' || status);
    -- RECEIVE_MESSAGE status is: 0

    status := DBMS_PIPE.NEXT_ITEM_TYPE;
    DBMS_OUTPUT.PUT_LINE('NEXT_ITEM_TYPE is: ' || status);
    -- NEXT_ITEM_TYPE is: 9

    DBMS_PIPE.UNPACK_MESSAGE(str);
    DBMS_OUTPUT.PUT_LINE('VARCHAR2 item is: ' || str);
    -- VARCHAR2 item is: Character data
END;

PACK_MESSAGE procedures

Adds a data item to the local message buffer. Call this once for each item to include in the message, then send the buffer with SEND_MESSAGE.

Syntax

DBMS_PIPE.PACK_MESSAGE(item IN VARCHAR2);
DBMS_PIPE.PACK_MESSAGE(item IN NUMBER);
DBMS_PIPE.PACK_MESSAGE(item IN DATE);
DBMS_PIPE.PACK_MESSAGE_RAW(item IN RAW);

Parameters

Parameter	Description
`item`	The data item to add to the local message buffer

Example

This example packs a VARCHAR2 and a NUMBER into the buffer and sends both to a pipe. To see how to receive and unpack this message, see the UNPACK_MESSAGE example.

DECLARE
    str    VARCHAR2 := 'Character data';
    num    NUMBER   := 1024;
    status INTEGER;
BEGIN
    DBMS_PIPE.PACK_MESSAGE(str);
    DBMS_PIPE.PACK_MESSAGE(num);

    status := DBMS_PIPE.SEND_MESSAGE('messages');
END;

UNPACK_MESSAGE procedures

Retrieves the next data item from the local message buffer. Call RECEIVE_MESSAGE first to load the buffer, then call UNPACK_MESSAGE once per item.

Syntax

DBMS_PIPE.UNPACK_MESSAGE(item OUT VARCHAR2);
DBMS_PIPE.UNPACK_MESSAGE(item OUT NUMBER);
DBMS_PIPE.UNPACK_MESSAGE(item OUT DATE);
DBMS_PIPE.UNPACK_MESSAGE_RAW(item OUT RAW);

Parameters

Parameter	Description
`item`	Variable to receive the next data item from the local message buffer

Example

This example receives a message from the messages pipe and unpacks the two items packed in the PACK_MESSAGE example.

DECLARE
    str    VARCHAR2;
    num    NUMBER;
    status INTEGER;
BEGIN
    status := DBMS_PIPE.RECEIVE_MESSAGE('messages');
    DBMS_OUTPUT.PUT_LINE('RECEIVE_MESSAGE status is: ' || status);
    -- RECEIVE_MESSAGE status is: 0

    DBMS_PIPE.UNPACK_MESSAGE(str);
    DBMS_OUTPUT.PUT_LINE('VARCHAR2 item is: ' || str);
    -- VARCHAR2 item is: Character data

    DBMS_PIPE.UNPACK_MESSAGE(num);
    DBMS_OUTPUT.PUT_LINE('NUMBER item is: ' || num);
    -- NUMBER item is: 1024
END;

PURGE procedure

Discards all pending messages in a named pipe.

Syntax

DBMS_PIPE.PURGE(pipename IN VARCHAR2);

Parameters

Parameter	Description
`pipename`	Name of the pipe whose messages to discard

Example

CALL DBMS_PIPE.PURGE('messages');

RECEIVE_MESSAGE function

Copies the next message from a named pipe into the local message buffer. After this call succeeds, use UNPACK_MESSAGE to read the individual items.

Syntax

DBMS_PIPE.RECEIVE_MESSAGE(
   pipename IN VARCHAR2,
   timeout  IN INTEGER DEFAULT 86400
) RETURN INTEGER;

Parameters

Parameter	Description
`pipename`	Name of the pipe to read from
`timeout`	(Optional) How long to wait for a message, in seconds. Default: `86400`

Return values

Return value	Description
INTEGER	The returned status code

Example

This example packs three VARCHAR2 items into a message, sends them to the messages pipe, then receives and unpacks all three in a loop.

DECLARE
    varchar VARCHAR2 := 'Character data';
    result  VARCHAR2;
    status  INTEGER;
    idx     INTEGER;
BEGIN
    DBMS_PIPE.PACK_MESSAGE(varchar);
    DBMS_PIPE.PACK_MESSAGE(varchar);
    DBMS_PIPE.PACK_MESSAGE(varchar);

    status := DBMS_PIPE.SEND_MESSAGE('messages');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status is: ' || status);
    -- SEND_MESSAGE status is: 0

    status := DBMS_PIPE.RECEIVE_MESSAGE('messages');

    FOR idx IN 1..3 LOOP
        DBMS_PIPE.UNPACK_MESSAGE(result);
        DBMS_OUTPUT.PUT_LINE(result);
        -- Character data
        -- Character data
        -- Character data
    END LOOP;
END;

REMOVE_PIPE function

Removes a named pipe. Use this to clean up explicit pipes when they are no longer needed.

Syntax

DBMS_PIPE.REMOVE_PIPE(pipename IN VARCHAR2) RETURN INTEGER;

Parameters

Parameter	Description
`pipename`	Name of the pipe to remove

Return values

Return value	Description
`0`	Pipe removed successfully

Example

SELECT DBMS_PIPE.REMOVE_PIPE('messages') FROM DUAL;

Expected output:

 remove_pipe
-------------
           0
(1 row)

RESET_BUFFER procedure

Clears the local message buffer. Use this to discard items that have been packed with PACK_MESSAGE but not yet sent.

Syntax

DBMS_PIPE.RESET_BUFFER;

Example

CALL DBMS_PIPE.RESET_BUFFER;

SEND_MESSAGE function

Sends the contents of the local message buffer to a named pipe. Pack items into the buffer with PACK_MESSAGE before calling this function.

Syntax

DBMS_PIPE.SEND_MESSAGE(
    pipename    IN VARCHAR2,
    timeout     IN INTEGER DEFAULT 86400,
    maxpipesize IN INTEGER DEFAULT 8192
) RETURN INTEGER;

Parameters

Parameter	Description
`pipename`	Name of the pipe to send to
`timeout`	(Optional) The waiting period, in seconds. Default: `86400`
`maxpipesize`	(Optional) Maximum capacity of the pipe, in bytes. Default: `8192`

Return values

Return value	Description
INTEGER	The returned status code

Example

This example sends three VARCHAR2 items to the messages pipe, then receives and unpacks them.

DECLARE
    varchar VARCHAR2 := 'Character data';
    result  VARCHAR2;
    status  INTEGER;
    idx     INTEGER;
BEGIN
    DBMS_PIPE.PACK_MESSAGE(varchar);
    DBMS_PIPE.PACK_MESSAGE(varchar);
    DBMS_PIPE.PACK_MESSAGE(varchar);

    status := DBMS_PIPE.SEND_MESSAGE('messages');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status is: ' || status);
    -- SEND_MESSAGE status is: 0

    status := DBMS_PIPE.RECEIVE_MESSAGE('messages');

    FOR idx IN 1..3 LOOP
        DBMS_PIPE.UNPACK_MESSAGE(result);
        DBMS_OUTPUT.PUT_LINE(result);
        -- Character data
        -- Character data
        -- Character data
    END LOOP;
END;

UNIQUE_SESSION_NAME function

Returns a name that is unique to the current session within the cluster. Use this to create session-specific pipe names that avoid collisions.

Syntax

DBMS_PIPE.UNIQUE_SESSION_NAME RETURN VARCHAR2;

Return values

Return value	Description
VARCHAR2	A unique identifier for the current session

Example

SELECT DBMS_PIPE.UNIQUE_SESSION_NAME() FROM DUAL;

Expected output:

  unique_session_name
-----------------------
 POLARDB$PIPE$12$57884
(1 row)