-- PL/SQL Job control modelled on Unix: submit one or more tasks, wait for them to complete.
-- Uses DBMS_JOB to submit tasks, DBMS_ALERT for jobs to pass back started/completed/failed messages.
-- The DBMS_ALERT mechanism is hidden from the submit/wait interface.
--
-- Object-based approach allows you to declare a JOB object then call myjob.SUBMIT() etc.
-- (See demo block for examples.)
--
-- Prerequisites: Execute permission on SYS.DBMS_ALERT.
--
-- William Robertson 2004, www.williamrobertson.net

-- Package (spec only) for constants used in type body:
CREATE OR REPLACE PACKAGE job_pkg
AS
	-- k_spec_version       CONSTANT VARCHAR2(100) := FORMAT_VERSION_STRING('$RCSfile$ $Revision$');
	-- version              CONSTANT VARCHAR2(100) := k_spec_version;

	k_error_code         CONSTANT PLS_INTEGER := -20000;
	k_delimiter          CONSTANT VARCHAR2(3) := CHR(0);  -- Delimiter for packing messages
	k_job_queue_interval CONSTANT INTERVAL DAY(0) TO SECOND(0) DEFAULT INTERVAL '5' SECOND;

	-- Job statuses:
	k_submitted          CONSTANT VARCHAR2(30) := 'Submitted';
	k_not_submitted      CONSTANT VARCHAR2(30) := 'Not submitted';
	k_waiting_confirm    CONSTANT VARCHAR2(30) := 'Awaiting start confirmation';
	k_submit_failed      CONSTANT VARCHAR2(30) := 'Submit failed';
	k_in_progress        CONSTANT VARCHAR2(30) := 'In Progress';
	k_wait_timed_out     CONSTANT VARCHAR2(30) := 'Wait timeout';
	k_failed             CONSTANT VARCHAR2(30) := 'Abnormal Termination';
	k_completed_ok       CONSTANT VARCHAR2(30) := 'Completed Successfully';
END job_pkg;
/

show errors

-- JOB_OT object type:
-- Submit one or more tasks, wait for them to complete.
-- Uses DBMS_JOB to submit tasks, DBMS_ALERT to pass back started/completed/failed messages.
-- The DBMS_ALERT mechanism is hidden from the submit/wait interface.
--
-- Object-based approach allows you to declare a new JOB_OT object then call myjob.SUBMIT() etc.

CREATE SEQUENCE job_seq START WITH 1 MAXVALUE 999 CYCLE;

CREATE OR REPLACE TYPE job_ot AS OBJECT
( name            VARCHAR2(30)
, job_number      NUMBER             -- Number set by DBMS_JOB.SUBMIT
, handle          VARCHAR2(30)       -- Generated unique ID formed from name + generated sequence
, command         VARCHAR2(4000)     -- Command to execute
, description     VARCHAR2(500)      -- Text description
, message         VARCHAR2(1000)     -- Message returned from execution; error message if failed
, wait_timeout    INTERVAL DAY TO SECOND(0)
, created_time    TIMESTAMP
, submitted_time  TIMESTAMP
, started_time    TIMESTAMP
, completed_time  TIMESTAMP
, run_status      VARCHAR2(50)       -- In progress, failed etc

, MEMBER PROCEDURE submit
, MEMBER PROCEDURE wait
	( p_timeout INTERVAL DAY TO SECOND DEFAULT NULL
	, p_alertname VARCHAR2 DEFAULT NULL )  -- What to wait for: defaults to self.handle, above
, CONSTRUCTOR FUNCTION job_ot
	( p_name             VARCHAR2
	, p_command          VARCHAR2
	, p_description      VARCHAR2 DEFAULT NULL
	, p_timeout          INTERVAL DAY TO SECOND DEFAULT INTERVAL '1' MINUTE )
	RETURN SELF AS RESULT
, STATIC PROCEDURE execute
	( p_command          VARCHAR2   -- What to execute
	, p_receipt_alert    VARCHAR2   -- Signal name for confirming receipt to submitting procedure
	, p_completed_alert  VARCHAR2   -- Signal name for confirming job completion to submitting procedure
	, p_invoking_session VARCHAR2 ) -- Session ID of submitting process
, MEMBER PROCEDURE print
)
/

show errors

CREATE OR REPLACE TYPE BODY job_ot AS
	CONSTRUCTOR FUNCTION job_ot
		( p_name          VARCHAR2
		, p_command       VARCHAR2
		, p_description   VARCHAR2 DEFAULT NULL
		, p_timeout       INTERVAL DAY TO SECOND DEFAULT INTERVAL '1' MINUTE )
		RETURN SELF AS RESULT
	IS
		v_seq  PLS_INTEGER;
		v_handle VARCHAR2(3000);  -- Bigger than required to ensure assignment does not fail
	BEGIN
		SELECT job_seq.nextval
		INTO   v_seq
		FROM   dual;

		v_handle := SUBSTR(NVL(UPPER(p_name),'ALERT') || '_' || v_seq,1,30);

		-- "<v_name>_RECEIPT" must not be longer than 35 characters:
		IF LENGTH(v_handle) > 27 THEN
			v_handle := SUBSTR(UPPER(p_name),1,14) || '_' || v_seq;
		END IF;

		SELF.name := p_name;
		SELF.handle := v_handle;
		SELF.command := p_command;
		SELF.description := p_description;
		SELF.wait_timeout := NVL(p_timeout,INTERVAL '1' MINUTE);
		SELF.created_time := SYSTIMESTAMP;
		SELF.run_status := job_pkg.k_not_submitted;
		RETURN;
	END;


	MEMBER PROCEDURE submit
	IS
		PRAGMA AUTONOMOUS_TRANSACTION;

		k_session_id CONSTANT VARCHAR2(30) := DBMS_SESSION.UNIQUE_SESSION_ID;
		k_wait_timeout INTERVAL DAY(0) TO SECOND(0) := job_pkg.k_job_queue_interval * 2;

		CURSOR c_jobs(cp_command VARCHAR2 := SELF.command)
		IS
			SELECT job
			FROM   user_jobs
			WHERE  what = cp_command;

		v_command VARCHAR2(5000);

		v_job_number   PLS_INTEGER;
		v_message      VARCHAR2(1000);
		v_job_status   PLS_INTEGER;

		v_receipt_name VARCHAR2(30);
	BEGIN
		IF SELF.command IS NULL THEN
			RAISE_APPLICATION_ERROR
			( job_pkg.k_error_code
			, 'Command must be specified' );
		ELSIF SELF.handle IS NULL THEN
			RAISE_APPLICATION_ERROR
			( job_pkg.k_error_code
			, 'Job handle must be specified' );
		END IF;

		v_receipt_name := SELF.handle || '_RECEIPT';

		v_command :=
			'JOB_OT.EXECUTE(''BEGIN '
			|| REPLACE(RTRIM(SELF.command,'; '),'''','''''')
			|| '; END;'', '''
			|| v_receipt_name || ''', '''
			|| SELF.handle
			|| ''','''
			|| k_session_id || ''');';

		-- In case same command already in job queue, remove any jobs for same command:
		FOR r_job IN c_jobs(SELF.command)
		LOOP
			DBMS_JOB.REMOVE(r_job.job);
		END LOOP;

		-- Register interest in both receipt and completion alerts.
		-- SUBMIT waits for receipt; calling procedure waits for completion.
		DBMS_ALERT.REGISTER(v_receipt_name);
		DBMS_ALERT.REGISTER(SELF.handle);
		COMMIT;

		submitted_time := SYSTIMESTAMP;

		BEGIN
			DBMS_JOB.SUBMIT
			( SELF.job_number  -- 'OUT' parameter: job number returned by DBMS_JOB.SUBMIT
			, v_command        -- in format 'JOB_OT.EXECUTE('BEGIN cmd; END;', receipt_alert, completed_alert);'
			, SYSDATE          -- First run
			, NULL );          -- Interval (none = no repeat)

			COMMIT;

			run_status := job_pkg.k_submitted;
		EXCEPTION
			WHEN OTHERS THEN
				RAISE_APPLICATION_ERROR
				( job_pkg.k_error_code
				, 'DBMS_JOB.SUBMIT failed'
				, TRUE );
		END;


		-- Wait for background job just submitted to confirm that it is ready to begin:
		-- Job should send initial confirmation before doing anything (picked up here) followed by
		-- completion signal on success/failure at end (picked up by calling procedure).

		-- Note that when a member procedure fails with an exception, changes to the object state
		-- are lost, e.g. RUN_STATUS will revert to 'Not submitted' - we cannot both raise an exception
		-- and change an attribute value.
		BEGIN
			-- WAIT() will raise exceptions on failure
			SELF.WAIT(k_wait_timeout, v_receipt_name);

			DBMS_ALERT.REMOVE(v_receipt_name);

			IF SELF.run_status = job_pkg.k_wait_timed_out THEN
				RAISE_APPLICATION_ERROR
				( job_pkg.k_error_code
				, 'Submitted task failed to start (no confirmation received after ' ||
				  LTRIM(TO_CHAR(k_wait_timeout),'+') || ')'
				, TRUE );
			END IF;
		END;

		started_time := SYSTIMESTAMP;
		run_status := job_pkg.k_in_progress;

		COMMIT;
	EXCEPTION
		WHEN others THEN
			run_status := job_pkg.k_submit_failed;

			RAISE_APPLICATION_ERROR
			( job_pkg.k_error_code
			, 'Failed to submit job ' || v_command
			, TRUE );
	END submit;


	MEMBER PROCEDURE wait
		( p_timeout INTERVAL DAY TO SECOND DEFAULT NULL
		, p_alertname VARCHAR2 DEFAULT NULL )  -- What to wait for: defaults to self.handle, above
	IS
		k_alertname  CONSTANT VARCHAR2(50) := NVL(p_alertname,SELF.handle);
		k_is_confirmation CONSTANT BOOLEAN := k_alertname = SELF.handle;
		k_timeout    CONSTANT INTERVAL DAY TO SECOND := NVL(p_timeout,SELF.wait_timeout);

		k_maxwait CONSTANT PLS_INTEGER :=
			NVL
			( EXTRACT(DAY FROM k_timeout) * 86400
			+ EXTRACT(HOUR FROM k_timeout) * 3600
			+ EXTRACT(MINUTE FROM k_timeout) * 60
			+ EXTRACT(SECOND FROM k_timeout)
			, DBMS_ALERT.MAXWAIT );

		v_delimiter_position PLS_INTEGER;
		v_message VARCHAR2(32767);
		v_dbms_alert_status INTEGER;
	BEGIN
		IF k_is_confirmation THEN
			run_status := job_pkg.k_waiting_confirm;
		END IF;

		DBMS_ALERT.WAITONE
		( k_alertname
		, v_message
		, v_dbms_alert_status
		, k_maxwait );

		DBMS_ALERT.REMOVE(k_alertname);

		v_delimiter_position := INSTR(v_message, job_pkg.k_delimiter);

		-- The returned message may be in the form "status[delimiter]message":
		-- if so, extract into run_status and message:
		IF v_delimiter_position > 0 THEN
			run_status := SUBSTR(v_message,1,v_delimiter_position -1);
			message :=    SUBSTR(v_message,v_delimiter_position + LENGTH(job_pkg.k_delimiter));
		ELSE
			message := v_message;
		END IF;

		IF v_dbms_alert_status = 1 THEN
			-- Check status returned by DBMS_ALERT: 1 = "timed out while waiting"
			SELF.run_status := job_pkg.k_wait_timed_out;

			SELF.message :=
			'Timeout after waiting ' || k_maxwait
			|| ' seconds for job ' || SELF.name || ': "'
			|| SUBSTR(SELF.command,1,80)
			|| CASE WHEN LENGTH(SELF.command) > 80 THEN '...' ELSE NULL END
			|| '"';

			-- Not good idea to have WAIT() fail, as then we lose any state changes that
			-- WAIT() made. Instead, the calling application should check thisjob.STATUS.
		ELSE
			IF k_is_confirmation THEN
				completed_time := SYSTIMESTAMP;
				message := NVL(message,job_pkg.k_completed_ok);
			END IF;

		END IF;
	END wait;


	STATIC PROCEDURE execute
		( p_command          VARCHAR2   -- What to execute
		, p_receipt_alert    VARCHAR2   -- Signal name for confirming receipt to submitting procedure
		, p_completed_alert  VARCHAR2   -- Signal name for confirming job completion to submitting procedure
		, p_invoking_session VARCHAR2 ) -- Session ID of submitting process
	IS
		PRAGMA AUTONOMOUS_TRANSACTION;

		v_status VARCHAR2(30) := job_pkg.k_in_progress;
		v_message VARCHAR2(1000) := job_pkg.k_in_progress;
	BEGIN
		DBMS_APPLICATION_INFO.SET_MODULE('JOB_OT.EXECUTE', p_command);

		-- Send initial message to confirm instruction received:
		-- (echo back the value of 'p_receipt_alert')
		DBMS_ALERT.SIGNAL(p_receipt_alert,v_status);
		COMMIT;

		BEGIN
			EXECUTE IMMEDIATE(p_command);

			v_message :=
			job_pkg.k_completed_ok ||
			job_pkg.k_delimiter ||
			job_pkg.k_completed_ok;
		EXCEPTION
			WHEN others THEN
				v_message :=
				job_pkg.k_failed ||
				job_pkg.k_delimiter ||
				SQLERRM;
		END;

		-- Send specified alert to inform waiting process that task has completed:
		DBMS_ALERT.SIGNAL(p_completed_alert,v_message);
		COMMIT;
	END execute;


	MEMBER PROCEDURE print
	IS
		PROCEDURE print_item
			( p_name VARCHAR2
			, p_value VARCHAR2 )
		IS
		BEGIN
			DBMS_OUTPUT.PUT_LINE(RPAD(p_name || ':', 15) || p_value);
		END print_item;
	BEGIN
		PRINT_ITEM('Job name', name);
		PRINT_ITEM('Command', command);
		PRINT_ITEM('Description', description);
		PRINT_ITEM('Message', message);
		PRINT_ITEM('Created', created_time);
		PRINT_ITEM('Submitted', submitted_time);
		PRINT_ITEM('Started', started_time);
		PRINT_ITEM('Completed', completed_time);
		PRINT_ITEM('Run status', run_status);
		DBMS_OUTPUT.PUT_LINE(CHR(9));
	END print;
END;
/

show errors

set serverout on size 100000

PROMPT
PROMPT Demo/test (allow 20 seconds for submit/wait/output cycle to complete):
PROMPT

set echo on

DECLARE
	-- Create some job objects:

	v_ok_job JOB_OT := NEW JOB_OT('SHOULD_WORK', 'NULL;', 'Background job test: successful job');

	v_invalid_job JOB_OT := NEW JOB_OT('SHOULD_FAIL', 'nosuchcommand', 'Background job test: handle failure');

	v_slow_job JOB_OT :=
	NEW JOB_OT
	( 'SHOULD_TIMEOUT'
	, 'DBMS_LOCK.SLEEP(20)'   -- User must have execute permission on SYS.DBMS_LOCK
	, 'Background job demo 2: trapping failure'
	, INTERVAL '5' SECOND );  -- Job will fail to complete within time allowed
BEGIN
	DBMS_OUTPUT.PUT_LINE('Defined the following jobs:' || CHR(10));

	v_ok_job.PRINT();         -- Display basic info about the job
	v_invalid_job.PRINT();
	v_slow_job.PRINT();

	DBMS_OUTPUT.NEW_LINE;
	DBMS_OUTPUT.PUT_LINE('Calling SUBMIT() method of each job in turn...' || CHR(10));

	v_ok_job.SUBMIT();        -- Submit in background
	v_invalid_job.SUBMIT();
	v_slow_job.SUBMIT();

	DBMS_OUTPUT.PUT_LINE('Jobs have now been submitted in background.');
	DBMS_OUTPUT.PUT_LINE('We can now get on with something else until we are ready to wait for them to complete.');

	DBMS_OUTPUT.PUT_LINE(CHR(10)||'...'||CHR(10));

	DBMS_OUTPUT.PUT_LINE('Calling WAIT() method of each job in turn...' || CHR(10));
	v_ok_job.WAIT();
	v_invalid_job.WAIT();
	v_slow_job.WAIT();

	DBMS_OUTPUT.PUT_LINE('Job details after all jobs have returned:' || CHR(10));
	v_ok_job.PRINT();
	v_invalid_job.PRINT();
	v_slow_job.PRINT();
END;
/


set echo off