I will first have to say, I am still trying to wrap my brain around all the components of the Service Broker and how they work together, so it's very likely I'm missing something obvious... it's just not obvious to me.
A little summary: We have a report server that has replicated transaction information from a POS system that we are trying to export in "real time" (the battle of real-time versus regular checks and exports via scheduled task and synchronous procedures has been fought and lost) to a web service that will run analytics and event based activities based upon the data in the transaction.
To this end, I created a Service Broker system for asynchronous web service consumption. Here is what I did (not necessarily in order, sorry):
CREATE XML SCHEMA COLLECTION dcmeta.[TriggerMessageSchemaCollection]
AS
N'<?xml version="1.0" encoding="utf-16"?>
<xs:schema id="TriggerMessageTypeSchema"
targetNamespace="http://tempuri.org/TriggerMessageTypeSchema.xsd"
attributeFormDefault="unqualified"
elementFormDefault="qualified"
xmlns="http://tempuri.org/TriggerMessageTypeSchema.xsd"
xmlns:mstns="http://tempuri.org/TriggerMessageTypeSchema.xsd"
xmlns:xs="http://www.w3.org/2001/XMLSchema"
>
<xs:element name="triggerMessage" nillable="false">
<xs:complexType>
<xs:sequence>
<xs:element type="xs:string" name="schema" nillable="false" minOccurs="1" maxOccurs="1"/>
<xs:element type="xs:string" name="object" nillable="false" minOccurs="1" maxOccurs="1"/>
<xs:element name="recordIds" nillable="false" minOccurs="1" maxOccurs="1">
<xs:complexType>
<xs:sequence>
<xs:element type="xs:integer" name="recordId" maxOccurs="unbounded" minOccurs="1"/>
</xs:sequence>
</xs:complexType>
</xs:element>
</xs:sequence>
</xs:complexType>
</xs:element>
</xs:schema>';
CREATE MESSAGE TYPE [TriggerMessage]
VALIDATION = VALID_XML WITH SCHEMA COLLECTION dcmeta.TriggerMessageSchemaCollection;
CREATE CONTRACT [TriggerContract]
(
TriggerMessage SENT BY INITIATOR
);
CREATE SERVICE [TriggerInitiationService]
ON QUEUE dcmeta.TriggerQueue
(
TriggerContract
);
CREATE SERVICE [TriggerTargetService]
ON QUEUE dcmeta.TriggerQueue
(
TriggerContract
);
CREATE PROCEDURE dcmeta.[spProcessTriggerData]
AS
SET NOCOUNT ON;
begin
declare
@message_type sysname,
@dialog uniqueidentifier,
@message_body xml,
@schema sysname,
@object sysname,
@workflow_id int;
while (1=1) begin -- forever loop
waitfor(
receive top(1)
@message_type = cast(message_type_name as sysname),
@message_body = cast(message_body as XML),
@dialog = conversation_handle
from
dcmeta.TriggerQueue
), timeout 500
if (@@ROWCOUNT = 0 or @message_body is null) break;
select
@schema = mb.n.value('schema[1]', 'sysname'),
@object = mb.n.value('object[1]', 'sysname'),
@workflow_id = ap.WorkflowID
from
@message_body.nodes('.') as mb(n)
inner join
dcmeta.ApiWorkFlows ap
on ap.[Schema] = mb.n.value('schema[1]', 'sysname')
and ap.[Object] = mb.n.value('object[1]', 'sysname');
if (@schema is null
or @object is null
or @workflow_id is null
or @workflow_id <= 0) break;
insert into dcmeta.ExportedRows
(
[Schema],
[Object],
[RowId],
[ScheduleWorkflowEventId]
)
select
@schema,
@object,
mb.n.value('.', 'numeric(17)'),
dbo.StartDcWorkflow(@workflow_id) -- this is the CLR UDF that calls the data collection webservice and returns a job id that can be tracked later
from
@message_body.nodes('//recordIds/recordId') mb(n);
end conversation @dialog;
end
end
CREATE QUEUE dcmeta.TriggerQueue
WITH STATUS=ON,
ACTIVATION(
PROCEDURE_NAME=dcmeta.spProcessTriggerData,
MAX_QUEUE_READERS=10,
EXECUTE AS SELF
);
With this infrastructure in place, I create a trigger on my transactions table like so:
CREATE TRIGGER [dbo].[DC_Insert]
ON [dbo].[transact]
AFTER INSERT
AS
BEGIN
SET NOCOUNT ON;
declare
@message_body xml;
select
@message_body = (
select
'dbo' as [schema],
'transact' as [object],
cast(
(select
*
from
(
select distinct
i.trans_no as recordId
from
inserted i
) as recordIds
for xml path(''),
root('recordIds')
) as xml)
for xml path('triggerMessage')
);
if @message_body is not null begin
declare @handle uniqueidentifier;
begin dialog conversation @handle
from service TriggerInitiationService
to service 'TriggerTargetService'
on contract TriggerContract
with encryption = off;
send on conversation @handle
message type TriggerMessage(@message_body);
end
END
To test this, I insert a row into my dbo.transact
table, and look for results popping up on the other side of the webservice, but nothing shows up. So, following some suggestions on figuring out what went wrong, I look up the following data:
SELECT * FROM sys.transmission_queue;
select * from sys.conversation_endpoints;
...which returns the following information:
conversation_handle to_service_name to_broker_instance from_service_name service_contract_name enqueue_time message_sequence_number message_type_name is_conversation_error is_end_of_dialog message_body transmission_status priority
37E6E8F2-A948-E411-A431-402CF4F3BB1A TriggerTargetService TriggerInitiationService TriggerContract 2014-09-30 13:59:10.043 0 TriggerMessage 0 0 0xFFFE3C0074007200690067006700650072004D006500730073006100670065003E003C0073006300680065006D0061003E00640062006F003C002F0073006300680065006D0061003E003C006F0062006A006500630074003E007400720061006E0073006100630074003C002F006F0062006A006500630074003E003C007200650063006F00720064004900640073003E003C007200650063006F0072006400490064003E00310032003300340035003600370038003C002F007200650063006F0072006400490064003E003C002F007200650063006F00720064004900640073003E003C002F0074007200690067006700650072004D006500730073006100670065003E00 Service Broker received an error message on this conversation. Service Broker will not transmit the message; it will be held until the application ends the conversation. 5
conversation_handle conversation_id is_initiator service_contract_id conversation_group_id service_id lifetime state state_desc far_service far_broker_instance principal_id far_principal_id outbound_session_key_identifier inbound_session_key_identifier security_timestamp dialog_timer send_sequence last_send_tran_id end_dialog_sequence receive_sequence receive_sequence_frag system_sequence first_out_of_order_sequence last_out_of_order_sequence last_out_of_order_frag is_system priority
DCF00852-A545-E411-A431-402CF4F3BB1A CFFA9BDB-41EC-48DE-B379-1AF7C2531551 1 2 DDF00852-A545-E411-A431-402CF4F3BB1A 2 2082-10-14 21:02:35.720 SO STARTED_OUTBOUND ExternalActivatorService SiriusSQL 4 -1 00000000-0000-0000-0000-000000000000 00000000-0000-0000-0000-000000000000 1900-01-01 00:00:00.000 1900-01-01 00:00:00.000 0 0x000000000000 -1 0 0 0 -1 0 0 1 5
37E6E8F2-A948-E411-A431-402CF4F3BB1A 2B4DAABB-A1FF-4ED3-961D-B4BCD0C5CC06 1 65541 38E6E8F2-A948-E411-A431-402CF4F3BB1A 65543 2082-10-18 17:13:16.997 ER ERROR TriggerTargetService 00E2A7ED-35EB-44DE-AB95-216799EA2BF4 1 -1 00000000-0000-0000-0000-000000000000 00000000-0000-0000-0000-000000000000 1900-01-01 00:00:00.000 1900-01-01 00:00:00.000 1 0xB41300000000 -1 0 0 0 -1 0 0 0 5
... the thing that I noticed immediately was the transmission_status
column saying the following:
Service Broker received an error message on this conversation. Service Broker will not transmit the message; it will be held until the application ends the conversation.
... but I don't know how to troubleshoot this further. What is the error? How can I localize it? How can I fix it? Any help pointing me in the right direction for answering any of these questions would be greatly appreciated.
FOR XML
generation query. Are you sure you need to tie the knot that tight?..validation = valid_xml
is usually enough. Having this bit fixed, everything else should work as it is.trustworthy
option set, and Service Broker should be enabled in it. And you might run into problems with theexecute as self
on the target queue - it might be better to tryas owner
first.