DBMS_PIPE - Developer Guide| Alibaba Cloud Documentation Center

The DBMS_PIPE package provides the capability to send messages through a pipe within or between sessions connected to the same database cluster.

The following table lists the functions and stored procedures that are available in the DBMS_PIPE package.

Table 1. DBMS_PIPE functions and stored procedures
Function/stored procedure	Return type	Description
CREATE PIPE(pipename [, maxpipesize ] [, private ])	INTEGER	Explicitly creates a private pipe if private is "true" (the default) or a public pipe if private is "false".
NEXT ITEM TYPE	INTEGER	Determines the data type of the next item in a received message.
PACK MESSAGE(item)	N/A	Places item in the local message buffer of the current session.
PURGE(pipename)	N/A	Removes unreceived messages from the specified pipe.
RECEIVE MESSAGE(pipename [, timeout ])	INTEGER	Retrieves a message from a specified pipe.
REMOVE PIPE(pipename)	INTEGER	Deletes an explicitly created pipe.
RESET BUFFER	N/A	Resets the local message buffer.
SEND MESSAGE(pipename [, timeout ] [, maxpipesize ])	INTEGER	Sends a message on a pipe.
UNIQUE SESSION NAME	VARCHAR2	Obtains a unique session name.
UNPACK MESSAGE(item OUT)	N/A	Retrieves the next data item from a message into a type-compatible variable, item.

Pipes are categorized as implicit or explicit. An implicit pipe is created if a reference is made to a pipe name that was not previously created by the CREATE_PIPE function. For example, if the SEND_MESSAGE function is executed using a non-existent pipe name, a new implicit pipe is created with that name. An explicit pipe is created using the CREATE_PIPE function with the first parameter specified. The first parameter specifies the pipe name for the new pipe.

Pipes are also categorized as private or public. A private pipe can only be accessed by the user who created the pipe. Even a superuser cannot access a private pipe that was created by another user. A public pipe can be accessed by any user who has access to the DBMS_PIPE package.

A public pipe can only be created by using the CREATE_PIPE function with the third parameter set to FALSE. The CREATE_PIPE function can be used to create a private pipe by setting the third parameter to TRUE or by omitting the third parameter. All implicit pipes are private.

The individual data items or message lines are first built in a local message buffer, unique to the current session. The PACK_MESSAGE stored procedure builds the message in the local message buffer of the current session. The SEND_MESSAGE function is then used to send the message through the pipe.

The receiving of a message involves the reverse operation. The RECEIVE_MESSAGE function is used to retrieve a message from the specified pipe. The message is written to the local message buffer of the current session. The UNPACK_MESSAGE stored procedure is then used to transfer the message data items from the message buffer to program variables. If a pipe contains multiple messages, RECEIVE_MESSAGE retrieves the messages in first-in-first-out (FIFO) order.

Each session maintains separate message buffers for messages created with the PACK_MESSAGE stored procedure and messages retrieved by the RECEIVE_MESSAGE function. The messages can be both built and received in the same session. However, if consecutive RECEIVE_MESSAGE calls are made, only the message from the last RECEIVE_MESSAGE call will be preserved in the local message buffer.

CREATE_PIPE

The CREATE_PIPE function creates an explicit public pipe or an explicit private pipe with a specified name.

status INTEGER CREATE_PIPE(pipename VARCHAR2
  [, maxpipesize INTEGER ] [, private BOOLEAN ])

Parameters


Parameter	Description
pipename	The name of the pipe.
maxpipesize	The maximum capacity of the pipe. Unit: byte. Default value: 8192.
private	To create a public pipe, you need to set this parameter to FALSE. To create a private pipe, you need to set this parameter to TRUE. Default value: TRUE.
status	The status code returned by the operation. 0 indicates successful creation.

Examples

The following example creates a private pipe named messages:

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

The following example creates a public pipeline named mailbox:

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

NEXT_ITEM_TYPE

The NEXT_ITEM_TYPE function returns an integer code identifying the data type of the next data item in a message that has been retrieved into the local message buffer of the current session. If an item is removed from the local message buffer by the UNPACK_MESSAGE stored procedure, the NEXT_ITEM_TYPE function returns the data type code for the next available item. If no more items exist in the message, the code 0 is returned.

typecode INTEGER NEXT_ITEM_TYPE

Parameters


Parameter	Description
typecode	A code that identifies the data type of the next data item. Table 2 lists the code of each data type.

Table 2. Data type codes of NEXT_ITEM_TYPE
Type code	Data type
0	No more data items
9	NUMBER
11	VARCHAR2
13	DATE
23	RAW

Note The type codes listed in the table are not compatible with Oracle databases. Oracle assigns a different numbering sequence to the data types.

The following example shows a pipe packed with a NUMBER item, a VARCHAR2 item, a DATE item, and a RAW item. A second anonymous block then uses the NEXT_ITEM_TYPE function to display the type code of each item.

DECLARE
    v_number        NUMBER := 123;
    v_varchar       VARCHAR2(20) := 'Character data';
    v_date          DATE := SYSDATE;
    v_raw           RAW(4) := '21222324';
    v_status        INTEGER;
BEGIN
    DBMS_PIPE.PACK_MESSAGE(v_number);
    DBMS_PIPE.PACK_MESSAGE(v_varchar);
    DBMS_PIPE.PACK_MESSAGE(v_date);
    DBMS_PIPE.PACK_MESSAGE(v_raw);
    v_status := DBMS_PIPE.SEND_MESSAGE('datatypes');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status: ' || v_status);
EXCEPTION
    WHEN OTHERS THEN
        DBMS_OUTPUT.PUT_LINE('SQLERRM: ' || SQLERRM);
        DBMS_OUTPUT.PUT_LINE('SQLCODE: ' || SQLCODE);
END;

SEND_MESSAGE status: 0

DECLARE
    v_number        NUMBER;
    v_varchar       VARCHAR2(20);
    v_date          DATE;
    v_timestamp     TIMESTAMP;
    v_raw           RAW(4);
    v_status        INTEGER;
BEGIN
    v_status := DBMS_PIPE.RECEIVE_MESSAGE('datatypes');
    DBMS_OUTPUT.PUT_LINE('RECEIVE_MESSAGE status: ' || v_status);
    DBMS_OUTPUT.PUT_LINE('----------------------------------');

    v_status := DBMS_PIPE.NEXT_ITEM_TYPE;
    DBMS_OUTPUT.PUT_LINE('NEXT_ITEM_TYPE: ' || v_status);
    DBMS_PIPE.UNPACK_MESSAGE(v_number);
    DBMS_OUTPUT.PUT_LINE('NUMBER Item   : ' || v_number);
    DBMS_OUTPUT.PUT_LINE('----------------------------------');
    v_status := DBMS_PIPE.NEXT_ITEM_TYPE;
    DBMS_OUTPUT.PUT_LINE('NEXT_ITEM_TYPE: ' || v_status);
    DBMS_PIPE.UNPACK_MESSAGE(v_varchar);
    DBMS_OUTPUT.PUT_LINE('VARCHAR2 Item : ' || v_varchar);
    DBMS_OUTPUT.PUT_LINE('----------------------------------');

    v_status := DBMS_PIPE.NEXT_ITEM_TYPE;
    DBMS_OUTPUT.PUT_LINE('NEXT_ITEM_TYPE: ' || v_status);
    DBMS_PIPE.UNPACK_MESSAGE(v_date);
    DBMS_OUTPUT.PUT_LINE('DATE Item     : ' || v_date);
    DBMS_OUTPUT.PUT_LINE('----------------------------------');

    v_status := DBMS_PIPE.NEXT_ITEM_TYPE;
    DBMS_OUTPUT.PUT_LINE('NEXT_ITEM_TYPE: ' || v_status);
    DBMS_PIPE.UNPACK_MESSAGE(v_raw);
    DBMS_OUTPUT.PUT_LINE('RAW Item      : ' || v_raw);
    DBMS_OUTPUT.PUT_LINE('----------------------------------');

    v_status := DBMS_PIPE.NEXT_ITEM_TYPE;
    DBMS_OUTPUT.PUT_LINE('NEXT_ITEM_TYPE: ' || v_status);
    DBMS_OUTPUT.PUT_LINE('---------------------------------');
EXCEPTION
    WHEN OTHERS THEN
        DBMS_OUTPUT.PUT_LINE('SQLERRM: ' || SQLERRM);
        DBMS_OUTPUT.PUT_LINE('SQLCODE: ' || SQLCODE);
END;

RECEIVE_MESSAGE status: 0
----------------------------------
NEXT_ITEM_TYPE: 9
NUMBER Item   : 123
----------------------------------
NEXT_ITEM_TYPE: 11
VARCHAR2 Item : Character data
----------------------------------
NEXT_ITEM_TYPE: 13
DATE Item     : 02-OCT-07 11:11:43
----------------------------------
NEXT_ITEM_TYPE: 23
RAW Item      : 21222324
----------------------------------
NEXT_ITEM_TYPE: 0

PACK_MESSAGE

The PACK_MESSAGE stored procedure places an item of data in the local message buffer of the current session. You must call the PACK_MESSAGE stored procedure at least once before issuing a SEND_MESSAGE call.

PACK_MESSAGE(item { DATE | NUMBER | VARCHAR2 | RAW })

After you retrieve the message by issuing a RECEIVE_MESSAGE call, you can use the UNPACK_MESSAGE stored procedure to obtain data items.

Parameters


Parameter	Description
item	An expression that is used to calculate the acceptable parameter data types. The calculated value is added to the local message buffer of the session.

PURGE

The PURGE stored procedure removes unreceived messages from a specified implicit pipe.

PURGE(pipename VARCHAR2)

You can use the REMOVE_PIPE function to delete an explicit pipe.

Parameters


Parameter	Description
pipename	The name of the pipe.

Examples

Two messages are sent on a pipe:

DECLARE
    v_status        INTEGER;
BEGIN
    DBMS_PIPE.PACK_MESSAGE('Message #1');
    v_status := DBMS_PIPE.SEND_MESSAGE('pipe');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status: ' || v_status);

    DBMS_PIPE.PACK_MESSAGE('Message #2');
    v_status := DBMS_PIPE.SEND_MESSAGE('pipe');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status: ' || v_status);
END;

SEND_MESSAGE status: 0
SEND_MESSAGE status: 0

Receive the first message and unpack it:

DECLARE
    v_item          VARCHAR2(80);
    v_status        INTEGER;
BEGIN
    v_status := DBMS_PIPE.RECEIVE_MESSAGE('pipe',1);
    DBMS_OUTPUT.PUT_LINE('RECEIVE_MESSAGE status: ' || v_status);
    DBMS_PIPE.UNPACK_MESSAGE(v_item);
    DBMS_OUTPUT.PUT_LINE('Item: ' || v_item);
END;

RECEIVE_MESSAGE status: 0
Item: Message #1

Purge the pipe:

EXEC DBMS_PIPE.PURGE('pipe');

The following code example shows an attempt to retrieve the next message. The RECEIVE_MESSAGE call returns status code 1, which indicates that a timeout occurs because no message is available.

DECLARE
    v_item          VARCHAR2(80);
    v_status        INTEGER;
BEGIN
    v_status := DBMS_PIPE.RECEIVE_MESSAGE('pipe',1);
    DBMS_OUTPUT.PUT_LINE('RECEIVE_MESSAGE status: ' || v_status);
END;

RECEIVE_MESSAGE status: 1

RECEIVE_MESSAGE

The RECEIVE_MESSAGE function retrieves a message from a specified pipe.

status INTEGER RECEIVE_MESSAGE(pipename VARCHAR2
  [, timeout INTEGER ])

Parameters

pipename

The name of the pipe.

timeout

The timeout period. Unit: second. Default value: 86400000 (1000 days).

Status

The status code returned by the operation.

The following table lists possible status codes.


Status code	Description
0	The operation is successful.
1	A timeout occurs.
2	The message is too large for the buffer.

REMOVE_PIPE

The REMOVE_PIPE function deletes an explicit private pipe or explicit public pipe.

status INTEGER REMOVE_PIPE(pipename VARCHAR2)

You can use the REMOVE_PIPE function to delete an explicit pipe, such as a pipe created by using the CREATE_PIPE function.

Parameters


Parameter	Description
pipename	The name of the pipe.
status	The status code returned by the operation. A status code of 0 is returned even if the specified pipe does not exist.

Examples

Two messages are sent on a pipe:

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

    DBMS_PIPE.PACK_MESSAGE('Message #1');
    v_status := DBMS_PIPE.SEND_MESSAGE('pipe');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status: ' || v_status);

    DBMS_PIPE.PACK_MESSAGE('Message #2');
    v_status := DBMS_PIPE.SEND_MESSAGE('pipe');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status: ' || v_status);
END;

CREATE_PIPE status : 0
SEND_MESSAGE status: 0
SEND_MESSAGE status: 0

Receive the first message and unpack it:

DECLARE
    v_item          VARCHAR2(80);
    v_status        INTEGER;
BEGIN
    v_status := DBMS_PIPE.RECEIVE_MESSAGE('pipe',1);
    DBMS_OUTPUT.PUT_LINE('RECEIVE_MESSAGE status: ' || v_status);
    DBMS_PIPE.UNPACK_MESSAGE(v_item);
    DBMS_OUTPUT.PUT_LINE('Item: ' || v_item);
END;

RECEIVE_MESSAGE status: 0
Item: Message #1

Remove the pipe:

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

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

The following code example shows an attempt to retrieve the next message. The RECEIVE_MESSAGE call returns status code 1, which indicates that a timeout occurs because the pipe has been deleted.

DECLARE
    v_item          VARCHAR2(80);
    v_status        INTEGER;
BEGIN
    v_status := DBMS_PIPE.RECEIVE_MESSAGE('pipe',1);
    DBMS_OUTPUT.PUT_LINE('RECEIVE_MESSAGE status: ' || v_status);
END;

RECEIVE_MESSAGE status: 1

RESET_BUFFER

The RESET_BUFFER stored procedure resets a pointer to the local message buffer back to the beginning of the buffer. This causes subsequent PACK_MESSAGE calls to overwrite any data items that existed in the message buffer prior to the RESET_BUFFER call.

RESET_BUFFER

Examples

A message to John is written to the local message buffer. You can call the RESET_BUFFER stored procedure to replace this message with a message to Bob. The message to Bob is sent on the pipe.

DECLARE
    v_status        INTEGER;
BEGIN
    DBMS_PIPE.PACK_MESSAGE('Hi, John');
    DBMS_PIPE.PACK_MESSAGE('Can you attend a meeting at 3:00, today?') ;
    DBMS_PIPE.PACK_MESSAGE('If not, is tomorrow at 8:30 ok with you?') ;
    DBMS_PIPE.RESET_BUFFER;
    DBMS_PIPE.PACK_MESSAGE('Hi, Bob');
    DBMS_PIPE.PACK_MESSAGE('Can you attend a meeting at 9:30, tomorrow?') ;
    v_status := DBMS_PIPE.SEND_MESSAGE('pipe');
    DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE status: ' || v_status);
END;

SEND_MESSAGE status: 0

The message to Bob is displayed in the received message.

DECLARE
    v_item          VARCHAR2(80);
    v_status        INTEGER;
BEGIN
    v_status := DBMS_PIPE.RECEIVE_MESSAGE('pipe',1);
    DBMS_OUTPUT.PUT_LINE('RECEIVE_MESSAGE status: ' || v_status);
    DBMS_PIPE.UNPACK_MESSAGE(v_item);
    DBMS_OUTPUT.PUT_LINE('Item: ' || v_item);
    DBMS_PIPE.UNPACK_MESSAGE(v_item);
    DBMS_OUTPUT.PUT_LINE('Item: ' || v_item);
END;

RECEIVE_MESSAGE status: 0
Item: Hi, Bob
Item: Can you attend a meeting at 9:30, tomorrow?

SEND_MESSAGE

The SEND_MESSAGE function sends a message from the local message buffer to the specified pipe.

status SEND_MESSAGE(pipename VARCHAR2 [, timeout INTEGER ]
  [, maxpipesize INTEGER ])

Parameters


Parameter	Description
pipename	The name of the pipe.
timeout	The timeout period. Unit: second. Default value: 86400000 (1000 days).
maxpipesize	The maximum capacity of the pipe. Unit: byte. Default value: 8192 bytes.
Status	The status code returned by the operation.

The following table lists possible status codes.

Table 3. Status codes of SEND_MESSAGE
Status code	Description
0	The operation is successful.
1	A timeout occurs.
3	The function is interrupted.

UNIQUE_SESSION_NAME

The UNIQUE_SESSION_NAME function returns a name that is unique to the current session.

name VARCHAR2 UNIQUE_SESSION_NAME

Parameters


Parameter	Description
name	A unique session name.

Examples

The following anonymous block retrieves and displays a unique session name.

DECLARE
    v_session       VARCHAR2(30);
BEGIN
    v_session := DBMS_PIPE.UNIQUE_SESSION_NAME;
    DBMS_OUTPUT.PUT_LINE('Session Name: ' || v_session);
END;

Session Name: PG$PIPE$5$2752

UNPACK_MESSAGE

The UNPACK_MESSAGE stored procedure copies the data items of a message from the local message buffer to a specified program variable. Before you use the UNPACK_MESSAGE stored procedure, you must place the message in the local message buffer by using the RECEIVE_MESSAGE function.

UNPACK_MESSAGE(item OUT { DATE | NUMBER | VARCHAR2 | RAW })

Parameters


Parameter	Description
item	A variable that receives a data item from the local message buffer. This variable must be compatible with the type of the data item.

Comprehensive example

The following example uses a pipe as a "mailbox". A series of stored procedures are used to create the mailbox, to add a multi-item message to the mailbox (up to three items), and to display the full contents of the mailbox. These stored procedures are enclosed in a package named mailbox.

CREATE OR REPLACE PACKAGE mailbox
IS
    PROCEDURE create_mailbox;
    PROCEDURE add_message (
        p_mailbox   VARCHAR2,
        p_item_1    VARCHAR2,
        p_item_2    VARCHAR2 DEFAULT 'END',
        p_item_3    VARCHAR2 DEFAULT 'END'
    );
    PROCEDURE empty_mailbox (
        p_mailbox   VARCHAR2,
        p_waittime  INTEGER DEFAULT 10
    );
END mailbox;

CREATE OR REPLACE PACKAGE BODY mailbox
IS
    PROCEDURE create_mailbox
    IS
        v_mailbox   VARCHAR2(30);
        v_status    INTEGER;
    BEGIN
        v_mailbox := DBMS_PIPE.UNIQUE_SESSION_NAME;
        v_status := DBMS_PIPE.CREATE_PIPE(v_mailbox,1000,FALSE);
        IF v_status = 0 THEN
            DBMS_OUTPUT.PUT_LINE('Created mailbox: ' || v_mailbox);
        ELSE
            DBMS_OUTPUT.PUT_LINE('CREATE_PIPE failed - status: ' ||
                v_status);
        END IF;
    END create_mailbox;

    PROCEDURE add_message (
        p_mailbox   VARCHAR2,
        p_item_1    VARCHAR2,
        p_item_2    VARCHAR2 DEFAULT 'END',
        p_item_3    VARCHAR2 DEFAULT 'END'
    )
    IS
        v_item_cnt  INTEGER := 0;
        v_status    INTEGER;
    BEGIN
        DBMS_PIPE.PACK_MESSAGE(p_item_1);
        v_item_cnt := 1;
        IF p_item_2 ! = 'END' THEN
            DBMS_PIPE.PACK_MESSAGE(p_item_2);
            v_item_cnt := v_item_cnt + 1;
        END IF;
        IF p_item_3 ! = 'END' THEN
            DBMS_PIPE.PACK_MESSAGE(p_item_3);
            v_item_cnt := v_item_cnt + 1;
        END IF;
        v_status := DBMS_PIPE.SEND_MESSAGE(p_mailbox);
        IF v_status = 0 THEN
            DBMS_OUTPUT.PUT_LINE('Added message with ' || v_item_cnt ||
                ' item(s) to mailbox ' || p_mailbox);
        ELSE
            DBMS_OUTPUT.PUT_LINE('SEND_MESSAGE in add_message failed - ' ||
                'status: ' || v_status);
        END IF;
    END add_message;

    PROCEDURE empty_mailbox (
        p_mailbox   VARCHAR2,
        p_waittime  INTEGER DEFAULT 10
    )
    IS
        v_msgno     INTEGER DEFAULT 0;
        v_itemno    INTEGER DEFAULT 0;
        v_item      VARCHAR2(100);
        v_status    INTEGER;
    BEGIN
        v_status := DBMS_PIPE.RECEIVE_MESSAGE(p_mailbox,p_waittime);
        WHILE v_status = 0 LOOP
            v_msgno := v_msgno + 1;
            DBMS_OUTPUT.PUT_LINE('****** Start message #' || v_msgno ||
                ' ******');
            BEGIN
                LOOP
                    v_status := DBMS_PIPE.NEXT_ITEM_TYPE;
                    EXIT WHEN v_status = 0;
                    DBMS_PIPE.UNPACK_MESSAGE(v_item);
                    v_itemno := v_itemno + 1;
                    DBMS_OUTPUT.PUT_LINE('Item #' || v_itemno || ': ' ||
                        v_item);
                END LOOP;
                DBMS_OUTPUT.PUT_LINE('******* End message #' || v_msgno ||
                    ' *******');
                DBMS_OUTPUT.PUT_LINE('*');
                v_itemno := 0;
                v_status := DBMS_PIPE.RECEIVE_MESSAGE(p_mailbox,1);
            END;
        END LOOP;
        DBMS_OUTPUT.PUT_LINE('Number of messages received: ' || v_msgno);
        v_status := DBMS_PIPE.REMOVE_PIPE(p_mailbox);
        IF v_status = 0 THEN
            DBMS_OUTPUT.PUT_LINE('Deleted mailbox ' || p_mailbox);
        ELSE
            DBMS_OUTPUT.PUT_LINE('Could not delete mailbox - status: '
                || v_status);
        END IF;
    END empty_mailbox;
END mailbox;

The following example demonstrates the execution of the stored procedures in mailbox. The first stored procedure creates a public pipe with a name generated by the UNIQUE_SESSION_NAME function.

EXEC mailbox.create_mailbox;

Created mailbox: PG$PIPE$13$3940

By using the mailbox name, a user in the same database with access to the mailbox and DBMS_PIPE packages can add messages.

EXEC mailbox.add_message('PG$PIPE$13$3940','Hi, John','Can you attend a meeting at 3:00, today?','-- Mary');

Added message with 3 item(s) to mailbox PG$PIPE$13$3940

EXEC mailbox.add_message('PG$PIPE$13$3940','Don''t forget to submit your report','Thanks,','-- Joe');

Added message with 3 item(s) to mailbox PG$PIPE$13$3940

The contents of the mailbox can be emptied.

EXEC mailbox.empty_mailbox('PG$PIPE$13$3940');

****** Start message #1 ******
Item #1: Hi, John
Item #2: Can you attend a meeting at 3:00, today?
Item #3: -- Mary
******* End message #1 *******
*
****** Start message #2 ******
Item #1: Don't forget to submit your report
Item #2: Thanks,
Item #3: Joe
******* End message #2 *******
*
Number of messages received: 2
Deleted mailbox PG$PIPE$13$3940