40 — Service Broker & Queuing

Producer                          Consumer
────────                          ────────
BEGIN DIALOG CONVERSATION         RECEIVE (from queue)
  → message type validation         → process message
  → route lookup                    → END CONVERSATION
  → target service queue               ↑
SEND ON conversation_handle       Internal Activation
END CONVERSATION                    (stored proc started
                                     automatically when
                                     messages arrive)

1. Message Types
2. Contracts           (references message types)
3. Queues              (no dependencies)
4. Services            (references queues + contracts)
5. Routes              (for remote services only)
6. Endpoints           (for remote services only)

-- Allow any binary/text content
CREATE MESSAGE TYPE [//MyApp/Request]
    VALIDATION = NONE;

-- Must be well-formed XML
CREATE MESSAGE TYPE [//MyApp/Response]
    VALIDATION = WELL_FORMED_XML;

-- Must validate against a registered XML schema collection
CREATE MESSAGE TYPE [//MyApp/OrderEvent]
    VALIDATION = VALID_XML WITH SCHEMA COLLECTION [dbo].[OrderEventSchema];

-- Built-in end-of-conversation type (always exists)
-- [http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog]
-- [http://schemas.microsoft.com/SQL/ServiceBroker/Error]

-- Only the initiator sends requests; only the target sends responses
CREATE CONTRACT [//MyApp/RequestResponse]
(
    [//MyApp/Request]   SENT BY INITIATOR,
    [//MyApp/Response]  SENT BY TARGET
);

-- Either side can send (fire-and-forget or bidirectional)
CREATE CONTRACT [//MyApp/EventContract]
(
    [//MyApp/OrderEvent] SENT BY ANY
);

CREATE QUEUE [dbo].[RequestQueue]
    WITH
        STATUS = ON,
        RETENTION = OFF,               -- keep sent messages in queue; OFF is default/recommended
        POISON_MESSAGE_HANDLING (STATUS = ON);  -- auto-disable after 5 consecutive rollbacks

-- Queue with activation
CREATE QUEUE [dbo].[RequestQueue]
    WITH
        STATUS = ON,
        POISON_MESSAGE_HANDLING (STATUS = ON),
        ACTIVATION (
            STATUS = ON,
            PROCEDURE_NAME = [dbo].[usp_ProcessRequest],
            MAX_QUEUE_READERS = 5,
            EXECUTE AS OWNER
        );

-- Target service (receives requests)
CREATE SERVICE [//MyApp/TargetService]
    ON QUEUE [dbo].[RequestQueue]
    ([//MyApp/RequestResponse]);    -- must list all contracts this service accepts

-- Initiator service (sends requests, receives responses)
CREATE SERVICE [//MyApp/InitiatorService]
    ON QUEUE [dbo].[ResponseQueue]; -- initiator service needs its own queue for responses

-- Full setup script for a simple request/response queue
USE [MyDatabase];
GO

-- 1. Message types
CREATE MESSAGE TYPE [//MyApp/Request]   VALIDATION = WELL_FORMED_XML;
CREATE MESSAGE TYPE [//MyApp/Response]  VALIDATION = WELL_FORMED_XML;
GO

-- 2. Contract
CREATE CONTRACT [//MyApp/RequestResponse]
(
    [//MyApp/Request]   SENT BY INITIATOR,
    [//MyApp/Response]  SENT BY TARGET
);
GO

-- 3. Queues
CREATE QUEUE [dbo].[RequestQueue]
    WITH STATUS = ON, POISON_MESSAGE_HANDLING (STATUS = ON);

CREATE QUEUE [dbo].[ResponseQueue]
    WITH STATUS = ON, POISON_MESSAGE_HANDLING (STATUS = ON);
GO

-- 4. Services
CREATE SERVICE [//MyApp/TargetService]
    ON QUEUE [dbo].[RequestQueue]
    ([//MyApp/RequestResponse]);

CREATE SERVICE [//MyApp/InitiatorService]
    ON QUEUE [dbo].[ResponseQueue];
GO

DECLARE @conversation_handle UNIQUEIDENTIFIER;
DECLARE @message_body        XML;

-- Every message must be sent within a BEGIN/END DIALOG CONVERSATION
BEGIN DIALOG CONVERSATION @conversation_handle
    FROM SERVICE [//MyApp/InitiatorService]
    TO   SERVICE '//MyApp/TargetService'   -- string, not identifier
    ON CONTRACT [//MyApp/RequestResponse]
    WITH ENCRYPTION = OFF;                  -- ON requires certificates + routes

-- Build the payload
SET @message_body = N'<Request><OrderId>42</OrderId></Request>';

-- Send (atomically with the surrounding user transaction)
;SEND ON CONVERSATION @conversation_handle
    MESSAGE TYPE [//MyApp/Request] (@message_body);

-- If fire-and-forget (no response expected), end the conversation immediately
-- END CONVERSATION @conversation_handle;

-- The SEND is part of your transaction — if the transaction rolls back,
-- the message is NOT delivered. This is the "transactional inbox" pattern.

DECLARE @conversation_handle UNIQUEIDENTIFIER;
DECLARE @message_type        SYSNAME;
DECLARE @message_body        VARBINARY(MAX);

-- RECEIVE is a blocking statement; WAITFOR adds a timeout
WAITFOR (
    RECEIVE TOP(1)
        @conversation_handle = conversation_handle,
        @message_type        = message_type_name,
        @message_body        = message_body
    FROM [dbo].[RequestQueue]
), TIMEOUT 5000;    -- milliseconds; 0 = non-blocking

IF @conversation_handle IS NOT NULL
BEGIN
    IF @message_type = '//MyApp/Request'
    BEGIN
        -- Process the message
        DECLARE @xml XML = CAST(@message_body AS XML);
        DECLARE @order_id INT = @xml.value('(/Request/OrderId)[1]', 'INT');

        -- ... do work ...

        -- Send response
        ;SEND ON CONVERSATION @conversation_handle
            MESSAGE TYPE [//MyApp/Response]
            (CAST(N'<Response><Status>OK</Status></Response>' AS VARBINARY(MAX)));

        -- End conversation when done
        END CONVERSATION @conversation_handle;
    END
    ELSE IF @message_type = 'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog'
    BEGIN
        -- Initiator has ended its side; clean up
        END CONVERSATION @conversation_handle;
    END
    ELSE IF @message_type = 'http://schemas.microsoft.com/SQL/ServiceBroker/Error'
    BEGIN
        DECLARE @error_xml XML = CAST(@message_body AS XML);
        DECLARE @error_code INT     = @error_xml.value('(/Error/Code)[1]', 'INT');
        DECLARE @error_desc NVARCHAR(4000) = @error_xml.value('(/Error/Description)[1]', 'NVARCHAR(4000)');
        -- log error
        END CONVERSATION @conversation_handle;
    END
END

DECLARE @messages TABLE
(
    conversation_handle  UNIQUEIDENTIFIER,
    message_type_name    SYSNAME,
    message_body         VARBINARY(MAX)
);

WAITFOR (
    RECEIVE TOP(100)         -- batch up to 100 messages
        conversation_handle,
        message_type_name,
        message_body
    FROM [dbo].[RequestQueue]
    INTO @messages
), TIMEOUT 1000;

-- Process all rows in @messages as a set

CREATE OR ALTER PROCEDURE [dbo].[usp_ProcessRequest]
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @conversation_handle UNIQUEIDENTIFIER;
    DECLARE @message_type        SYSNAME;
    DECLARE @message_body        VARBINARY(MAX);

    -- Keep looping until the queue is empty
    WHILE (1 = 1)
    BEGIN
        BEGIN TRANSACTION;

        BEGIN TRY
            WAITFOR (
                RECEIVE TOP(1)
                    @conversation_handle = conversation_handle,
                    @message_type        = message_type_name,
                    @message_body        = message_body
                FROM [dbo].[RequestQueue]
            ), TIMEOUT 1000;

            -- No message available — exit cleanly
            IF @conversation_handle IS NULL
            BEGIN
                ROLLBACK TRANSACTION;
                RETURN;
            END

            -- Dispatch by message type
            IF @message_type = '//MyApp/Request'
            BEGIN
                DECLARE @xml      XML  = CAST(@message_body AS XML);
                DECLARE @order_id INT  = @xml.value('(/Request/OrderId)[1]', 'INT');

                -- Business logic here
                INSERT INTO dbo.ProcessedOrders (OrderId, ProcessedAt)
                VALUES (@order_id, SYSDATETIME());

                ;SEND ON CONVERSATION @conversation_handle
                    MESSAGE TYPE [//MyApp/Response]
                    (CAST(N'<Response><Status>OK</Status></Response>' AS VARBINARY(MAX)));

                END CONVERSATION @conversation_handle;
            END
            ELSE IF @message_type IN (
                'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog',
                'http://schemas.microsoft.com/SQL/ServiceBroker/Error'
            )
            BEGIN
                END CONVERSATION @conversation_handle;
            END

            COMMIT TRANSACTION;
        END TRY
        BEGIN CATCH
            ROLLBACK TRANSACTION;
            -- Log error — do NOT re-raise; that kills the activation proc
            -- and SSB will re-activate on next message (creating retry logic)
            INSERT INTO dbo.BrokerErrors (ErrorTime, ErrorMessage, ErrorSeverity)
            VALUES (SYSDATETIME(), ERROR_MESSAGE(), ERROR_SEVERITY());
        END CATCH
    END
END;
GO

-- Attach activation to queue
ALTER QUEUE [dbo].[RequestQueue]
    WITH ACTIVATION (
        STATUS           = ON,
        PROCEDURE_NAME   = [dbo].[usp_ProcessRequest],
        MAX_QUEUE_READERS = 5,          -- max concurrent instances
        EXECUTE AS       OWNER          -- or SELF, or 'username'
    );

USE [MyDatabase];
GO

-- ============================================================
-- INFRASTRUCTURE
-- ============================================================
CREATE MESSAGE TYPE [//MyApp/EmailRequest]  VALIDATION = WELL_FORMED_XML;
CREATE CONTRACT    [//MyApp/EmailContract]  ([//MyApp/EmailRequest] SENT BY INITIATOR);
CREATE QUEUE       [dbo].[EmailQueue]       WITH STATUS = ON, POISON_MESSAGE_HANDLING (STATUS = ON);
CREATE SERVICE     [//MyApp/EmailSender]    ON QUEUE [dbo].[EmailQueue] ([//MyApp/EmailContract]);
CREATE SERVICE     [//MyApp/EmailClient]    ON QUEUE [dbo].[EmailQueue];  -- fire-and-forget; reuse same queue
GO

-- ============================================================
-- CONSUMER (activation procedure)
-- ============================================================
CREATE OR ALTER PROCEDURE [dbo].[usp_SendEmailFromQueue]
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @ch   UNIQUEIDENTIFIER;
    DECLARE @mt   SYSNAME;
    DECLARE @body VARBINARY(MAX);
    DECLARE @xml  XML;

    WHILE (1 = 1)
    BEGIN
        BEGIN TRANSACTION;
        BEGIN TRY
            WAITFOR (
                RECEIVE TOP(1)
                    @ch   = conversation_handle,
                    @mt   = message_type_name,
                    @body = message_body
                FROM [dbo].[EmailQueue]
            ), TIMEOUT 500;

            IF @ch IS NULL
            BEGIN
                ROLLBACK;
                RETURN;
            END

            IF @mt = '//MyApp/EmailRequest'
            BEGIN
                SET @xml = CAST(@body AS XML);

                EXEC msdb.dbo.sp_send_dbmail
                    @profile_name  = 'Default',
                    @recipients    = @xml.value('(/Email/To)[1]',      'NVARCHAR(500)'),
                    @subject       = @xml.value('(/Email/Subject)[1]', 'NVARCHAR(500)'),
                    @body          = @xml.value('(/Email/Body)[1]',     'NVARCHAR(MAX)'),
                    @body_format   = 'HTML';
            END

            IF @mt IN (
                'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog',
                'http://schemas.microsoft.com/SQL/ServiceBroker/Error'
            )
                END CONVERSATION @ch;

            COMMIT;
        END TRY
        BEGIN CATCH
            ROLLBACK;
        END CATCH
    END
END;
GO

ALTER QUEUE [dbo].[EmailQueue]
    WITH ACTIVATION (
        STATUS            = ON,
        PROCEDURE_NAME    = [dbo].[usp_SendEmailFromQueue],
        MAX_QUEUE_READERS = 3,
        EXECUTE AS        OWNER
    );
GO

-- ============================================================
-- PRODUCER — call this from any OLTP transaction
-- ============================================================
CREATE OR ALTER PROCEDURE [dbo].[usp_QueueEmail]
    @To      NVARCHAR(500),
    @Subject NVARCHAR(500),
    @Body    NVARCHAR(MAX)
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @ch  UNIQUEIDENTIFIER;
    DECLARE @msg XML;

    SET @msg = (
        SELECT
            @To      AS [To],
            @Subject AS [Subject],
            @Body    AS [Body]
        FOR XML PATH('Email'), TYPE
    );

    BEGIN DIALOG CONVERSATION @ch
        FROM SERVICE [//MyApp/EmailClient]
        TO   SERVICE '//MyApp/EmailSender'
        ON CONTRACT [//MyApp/EmailContract]
        WITH ENCRYPTION = OFF;

    ;SEND ON CONVERSATION @ch
        MESSAGE TYPE [//MyApp/EmailRequest]
        (CAST(@msg AS VARBINARY(MAX)));

    -- Fire-and-forget: end our side immediately
    END CONVERSATION @ch WITH CLEANUP;
END;
GO

-- Usage: called inside a user transaction; email is queued atomically
BEGIN TRANSACTION;
    INSERT INTO dbo.Orders (CustomerId, Total) VALUES (101, 500.00);
    EXEC dbo.usp_QueueEmail
        @To      = 'customer@example.com',
        @Subject = 'Order Confirmed',
        @Body    = '<h1>Your order has been placed.</h1>';
COMMIT;
-- If the INSERT fails and we ROLLBACK, the SEND is also rolled back.
-- Email will never be sent. This is exactly the desired behavior.

-- Configure external activation via event notification
CREATE EVENT NOTIFICATION [EN_EmailQueueActivation]
    ON QUEUE [dbo].[EmailQueue]
    FOR QUEUE_ACTIVATION
    TO SERVICE '//MyApp/ExternalActivator', 'current database';

-- Remote route: messages for //RemoteApp/Service go to broker at remote server
CREATE ROUTE [RemoteServiceRoute]
    WITH
        SERVICE_NAME         = '//RemoteApp/TargetService',
        BROKER_INSTANCE      = '6E5C3D9A-...',   -- remote database's service_broker_guid
        ADDRESS              = 'TCP://remoteserver:4022';

-- Local route (explicit, within same instance)
CREATE ROUTE [LocalRoute]
    WITH
        SERVICE_NAME = '//MyApp/TargetService',
        ADDRESS      = 'LOCAL';

-- Default catch-all route (implicit; exists as AutoCreatedLocal)
-- Sends unrouted messages to the local broker

-- SSB endpoint example
CREATE ENDPOINT [SSBEndpoint]
    STATE = STARTED
    AS TCP (LISTENER_PORT = 4022)
    FOR SERVICE_BROKER (AUTHENTICATION = WINDOWS);

-- Fan-out: one sender, multiple queues
CREATE OR ALTER PROCEDURE [dbo].[usp_PublishOrderEvent]
    @event_xml XML
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @ch UNIQUEIDENTIFIER;
    DECLARE @payload VARBINARY(MAX) = CAST(@event_xml AS VARBINARY(MAX));

    -- Send to each subscriber service
    BEGIN DIALOG CONVERSATION @ch
        FROM SERVICE [//MyApp/Publisher]
        TO SERVICE '//MyApp/SubscriberA'
        ON CONTRACT [//MyApp/EventContract]
        WITH ENCRYPTION = OFF;
    ;SEND ON CONVERSATION @ch MESSAGE TYPE [//MyApp/OrderEvent] (@payload);
    END CONVERSATION @ch WITH CLEANUP;

    BEGIN DIALOG CONVERSATION @ch
        FROM SERVICE [//MyApp/Publisher]
        TO SERVICE '//MyApp/SubscriberB'
        ON CONTRACT [//MyApp/EventContract]
        WITH ENCRYPTION = OFF;
    ;SEND ON CONVERSATION @ch MESSAGE TYPE [//MyApp/OrderEvent] (@payload);
    END CONVERSATION @ch WITH CLEANUP;
END;

-- Explicit conversation group: related conversations share a group
DECLARE @group_id UNIQUEIDENTIFIER = NEWID();

BEGIN DIALOG CONVERSATION @ch1
    FROM SERVICE [//MyApp/Workflow]
    TO SERVICE '//MyApp/Worker'
    ON CONTRACT [//MyApp/WorkflowContract]
    WITH RELATED_CONVERSATION_GROUP = @group_id,
         ENCRYPTION = OFF;

BEGIN DIALOG CONVERSATION @ch2
    FROM SERVICE [//MyApp/Workflow]
    TO SERVICE '//MyApp/Worker'
    ON CONTRACT [//MyApp/WorkflowContract]
    WITH RELATED_CONVERSATION_GROUP = @group_id,
         ENCRYPTION = OFF;
-- ch1 and ch2 are in the same group; RECEIVE on one locks the other

-- Check for disabled queues
SELECT name, is_receive_enabled, is_enqueue_enabled, is_activation_enabled
FROM sys.service_queues
WHERE is_receive_enabled = 0;

-- Re-enable after fixing the root cause
ALTER QUEUE [dbo].[RequestQueue] WITH STATUS = ON;

-- Poison message safe-removal pattern
BEGIN TRANSACTION;
RECEIVE TOP(1) conversation_handle, message_type_name, message_body
FROM [dbo].[RequestQueue];
-- Inspect, log, or dead-letter the message
-- Do NOT process it; just consume and discard
COMMIT;

-- In the catch block of the activation proc
BEGIN TRANSACTION;

;SEND ON CONVERSATION @ch
    -- Re-route to a dead-letter service for later analysis
    -- (requires a separate dead-letter service/queue setup)
    MESSAGE TYPE [//MyApp/DeadLetter]
    (@original_body);

END CONVERSATION @ch;
COMMIT;

-- Queue table
CREATE TABLE dbo.JobQueue
(
    JobId       INT           NOT NULL IDENTITY(1,1) PRIMARY KEY,
    Payload     NVARCHAR(MAX) NOT NULL,
    QueuedAt    DATETIME2     NOT NULL DEFAULT SYSDATETIME(),
    VisibleAfter DATETIME2    NOT NULL DEFAULT SYSDATETIME()
);

-- Producer
INSERT INTO dbo.JobQueue (Payload) VALUES ('{"type":"SendEmail","to":"x@y.com"}');

-- Consumer: dequeue one item atomically (no double-delivery)
DECLARE @job TABLE (JobId INT, Payload NVARCHAR(MAX));

DELETE TOP(1) FROM dbo.JobQueue
OUTPUT deleted.JobId, deleted.Payload INTO @job
WHERE VisibleAfter <= SYSDATETIME();

-- Process @job

CREATE TABLE dbo.JobQueue
(
    JobId        INT           NOT NULL IDENTITY(1,1) PRIMARY KEY,
    Payload      NVARCHAR(MAX) NOT NULL,
    LockedUntil  DATETIME2     NULL,          -- NULL = available
    LockedBy     UNIQUEIDENTIFIER NULL        -- worker ID
);

-- Dequeue with visibility timeout (claim for 5 minutes)
DECLARE @worker_id UNIQUEIDENTIFIER = NEWID();
DECLARE @job       TABLE (JobId INT, Payload NVARCHAR(MAX));

UPDATE TOP(1) dbo.JobQueue
SET
    LockedUntil = DATEADD(MINUTE, 5, SYSDATETIME()),
    LockedBy    = @worker_id
OUTPUT inserted.JobId, inserted.Payload INTO @job
WHERE LockedUntil IS NULL OR LockedUntil < SYSDATETIME();

-- After successful processing
DELETE FROM dbo.JobQueue WHERE JobId = (SELECT JobId FROM @job);
-- If processing fails, LockedUntil expires and another worker can claim it

-- SSB must be enabled for Query Notifications to work
ALTER DATABASE [MyDatabase] SET ENABLE_BROKER;

-- The query being monitored must qualify (SELECT only, schema-qualified,
-- no aggregates, no subqueries on base tables, no non-deterministic funcs)
-- Example qualifying query:
SELECT OrderId, Status FROM dbo.Orders WHERE CustomerId = 42;

// .NET SqlDependency usage
using SqlDependency dep = new SqlDependency(cmd);
dep.OnChange += (sender, e) =>
{
    // Re-execute the query; e.Info has reason (Insert/Update/Delete/Invalid/etc.)
};
cmd.ExecuteReader();

-- Queue depth (backlog per queue)
SELECT
    q.name                                                   AS queue_name,
    COUNT(*)                                                 AS message_count,
    SUM(DATALENGTH(msg.message_body))                        AS total_bytes
FROM sys.service_queues q
LEFT JOIN sys.transmission_queue msg ON 1=0   -- placeholder; see below
GROUP BY q.name;

-- Actual message count per queue (using internal catalog)
SELECT
    q.name          AS queue_name,
    sq.message_count
FROM sys.service_queues q
CROSS APPLY (
    SELECT COUNT(*) AS message_count
    FROM sys.dm_broker_queue_monitors m
    WHERE m.queue_id = q.object_id
) sq;

-- Better: use sys.dm_broker_queue_monitors
SELECT
    database_id,
    queue_id,
    state,               -- INACTIVE, NOTIFIED, RECEIVES_OCCURRING
    last_empty_rowset_time,
    last_activated_time
FROM sys.dm_broker_queue_monitors;

-- Transmission queue: messages pending routing/delivery to remote services
SELECT
    to_service_name,
    to_broker_instance,
    message_type_name,
    transmission_status,   -- NULL = successful; error text if failed
    enqueue_time,
    message_id
FROM sys.transmission_queue
ORDER BY enqueue_time;

-- Active conversations
SELECT
    c.conversation_id,
    c.state_desc,        -- STARTED_OUTBOUND, CONVERSING, DISCONNECTED_INBOUND, etc.
    c.is_initiator,
    c.far_service,
    c.lifetime,
    c.security_timestamp
FROM sys.conversation_endpoints c;

-- Activation state
SELECT
    q.name,
    q.activation_procedure,
    q.max_readers,
    q.is_activation_enabled,
    q.is_receive_enabled,
    q.is_enqueue_enabled
FROM sys.service_queues q;

-- Broker configuration
SELECT
    name,
    is_broker_enabled,
    service_broker_guid
FROM sys.databases
WHERE name = DB_NAME();

-- Orphaned conversation endpoints (conversations never properly ended)
-- These accumulate over time if END CONVERSATION is missing
SELECT COUNT(*) AS orphaned
FROM sys.conversation_endpoints
WHERE state_desc NOT IN ('CONVERSING', 'STARTED_OUTBOUND')
AND   lifetime < DATEADD(DAY, -7, GETUTCDATE());

-- Force-close orphaned conversations (use with caution)
DECLARE @ch UNIQUEIDENTIFIER;
DECLARE cur CURSOR FAST_FORWARD FOR
    SELECT conversation_handle
    FROM sys.conversation_endpoints
    WHERE state_desc = 'DISCONNECTED_INBOUND'
    AND   lifetime < DATEADD(DAY, -1, GETUTCDATE());

OPEN cur;
FETCH NEXT FROM cur INTO @ch;
WHILE @@FETCH_STATUS = 0
BEGIN
    END CONVERSATION @ch WITH CLEANUP;
    FETCH NEXT FROM cur INTO @ch;
END
CLOSE cur;
DEALLOCATE cur;

-- Check for accumulated sys_transmission_queue entries
SELECT COUNT(*) FROM sys.transmission_queue;

-- Disable SSB for migration/maintenance
ALTER DATABASE [MyDatabase] SET DISABLE_BROKER;
-- Re-enable (assigns a new service_broker_guid — breaks existing remote routes)
ALTER DATABASE [MyDatabase] SET ENABLE_BROKER;
-- Re-enable with NEW_BROKER (new guid) or ERROR_BROKER_CONVERSATIONS (errors existing)
ALTER DATABASE [MyDatabase] SET NEW_BROKER;