src/openSourceC.DotNetLibrary.Data.SqlClient/Data/SqlClient/SqlDbFactory.cs
/// <summary>
///		Begin execution of a listener for a <see cref="T:SqlNotificationRequest"/> that will
///		execute on a different <see cref="T:SqlDbFactory"/>.
/// </summary>
/// <param name="queueName">The Service Broker queue name.</param>
/// <param name="timeout">The number of seconds to wait for the T-SQL RECEIVE statement to
///		return at least 1 message, or -1 to wait indefinitely.</param>
///	<param name="exceptionCallback">The callback method for exceptions thrown during the
///		asynchronous execution period of this call.</param>
/// <returns>
///		An <see cref="T:IAsyncResult"/> that can be used to poll or wait for results, or
///		both; this value is also needed when invoking <see cref="T:EndExecuteReader"/>,
///		which returns a <see cref="T:SqlDataReader"/> instance which can be used to retrieve
///		the returned rows.
/// </returns>
Begin execution of a listener for a  that will
execute on a different .
Begin execution of a listener for a that will execute on a different .
public IAsyncResult BeginNotificationListener(string queueName, int timeout, Action<Exception> exceptionCallback)
		{
			if (exceptionCallback == null) { throw new ArgumentNullException(nameof(exceptionCallback)); }
			EnsureServiceBroker();
			if (NotificationReceived == null)
			{
				throw new InvalidOperationException($"The NotificationReceived event has no subscribers.");
			}
			int commandTimeoutExtension = 120;
			StringBuilder commandText = new StringBuilder();
			if (_conversationHandle.HasValue)
			{
				commandText.AppendLine($"END CONVERSATION @ConversationHandle;");
			}
			commandText.AppendLine($"WAITFOR (RECEIVE TOP(1)");
			commandText.AppendLine($"\t[status],");
			commandText.AppendLine($"\t[priority],");
			commandText.AppendLine($"\t[queuing_order],");
			commandText.AppendLine($"\t[conversation_group_id],");
			commandText.AppendLine($"\t[conversation_handle],");
			commandText.AppendLine($"\t[message_sequence_number],");
			commandText.AppendLine($"\t[service_name],");
			commandText.AppendLine($"\t[service_id],");
			commandText.AppendLine($"\t[service_contract_name],");
			commandText.AppendLine($"\t[service_contract_id],");
			commandText.AppendLine($"\t[message_type_name],");
			commandText.AppendLine($"\t[message_type_id],");
			commandText.AppendLine($"\t[validation],");
			commandText.AppendLine($"\tCAST([message_body] AS XML) [message_body]");
			commandText.AppendLine($"FROM");
			commandText.AppendLine($"\t{queueName}");
			commandText.AppendLine($"), TIMEOUT @TimeoutMilliseconds;");
			SqlDbFactoryCommand cmd = CreateTextCommand(commandText.ToString());
			if (_conversationHandle.HasValue)
			{
				cmd.Params.AddGuid("ConversationHandle", false, _conversationHandle);
			}
			cmd.Params.AddInt64("TimeoutMilliseconds", false, (timeout == -1 ? timeout : timeout * 1000));
			cmd.CommandTimeout = (
				timeout == -1
				? 0 
				: timeout >= int.MaxValue - commandTimeoutExtension
				? int.MaxValue
				: timeout + commandTimeoutExtension
			);
			AsyncCallback callBack = new AsyncCallback(endNotificationListener);
			_conversationHandle = null;
			return cmd.BeginExecuteReader(callBack, cmd);
			void endNotificationListener(IAsyncResult ar)
			{
				try
				{
#if DIAGNOSTICS
					Debug.WriteLine($"Listener Reader: {queueName}");
#endif
					SqlNotificationReceiveEventArgs eventArgs;
					using (cmd)
					{
						try
						{
							using (SqlDataReader reader = cmd.EndExecuteReader(ar))
							{
								if (reader.Read())
								{
									eventArgs = new SqlNotificationReceiveEventArgs();
									eventArgs.Status = reader.GetByte("status");
									eventArgs.Priority = reader.GetByte("priority");
									eventArgs.QueuingOrder = reader.GetInt64("queuing_order");
									eventArgs.ConversationGroupId = reader.GetGuid("conversation_group_id");
									eventArgs.ConversationHandle = reader.GetGuid("conversation_handle");
									eventArgs.MessageSequenceNumber = reader.GetInt64("message_sequence_number");
									eventArgs.ServiceName = reader.GetString("service_name");
									eventArgs.ServiceId = reader.GetInt32("service_id");
									eventArgs.ServiceContractName = reader.GetString("service_contract_name");
									eventArgs.ServiceContractId = reader.GetInt32("service_contract_id");
									eventArgs.MessageTypeName = reader.GetString("message_type_name");
									eventArgs.MessageTypeId = reader.GetInt32("message_type_id");
									eventArgs.Validation = reader.GetString("validation");
									eventArgs.MessageBody = reader.GetStringNullSafe("message_body");
									_conversationHandle = eventArgs.ConversationHandle;
#if DIAGNOSTICS
									Debug.WriteLine($"Listener Reader: ConversationHandle={obj.ConversationHandle}, MessageTypeName={obj.MessageTypeName}, MessageBody={obj.MessageBody}");
#endif
								}
								else
								{
									eventArgs = null;
								}
							}
						}
						catch (Exception ex)
						{
							throw new DbCommandException(cmd.Command, ex);
						}
					}
					if (eventArgs != null)
					{
						NotificationReceived?.Invoke(this, eventArgs);
					}
				}
				catch (Exception ex)
				{
					exceptionCallback(ex);
				}
			}
		}
public IAsyncResult BeginNotificationListener ( string queueName , int timeout , Action < Exception > exceptionCallback ) { if ( exceptionCallback == null ) { throw new ArgumentNullException ( nameof ( exceptionCallback ) ) ; } EnsureServiceBroker ( ) ; if ( NotificationReceived == null ) { throw new InvalidOperationException ( $" The NotificationReceived event has no subscribers. " ) ; } int commandTimeoutExtension = 120 ; StringBuilder commandText = new StringBuilder ( ) ; if ( _conversationHandle . HasValue ) { commandText . AppendLine ( $" END CONVERSATION @ConversationHandle; " ) ; } commandText . AppendLine ( $" WAITFOR (RECEIVE TOP(1) " ) ; commandText . AppendLine ( $" \t [status], " ) ; commandText . AppendLine ( $" \t [priority], " ) ; commandText . AppendLine ( $" \t [queuing_order], " ) ; commandText . AppendLine ( $" \t [conversation_group_id], " ) ; commandText . AppendLine ( $" \t [conversation_handle], " ) ; commandText . AppendLine ( $" \t [message_sequence_number], " ) ; commandText . AppendLine ( $" \t [service_name], " ) ; commandText . AppendLine ( $" \t [service_id], " ) ; commandText . AppendLine ( $" \t [service_contract_name], " ) ; commandText . AppendLine ( $" \t [service_contract_id], " ) ; commandText . AppendLine ( $" \t [message_type_name], " ) ; commandText . AppendLine ( $" \t [message_type_id], " ) ; commandText . AppendLine ( $" \t [validation], " ) ; commandText . AppendLine ( $" \t CAST([message_body] AS XML) [message_body] " ) ; commandText . AppendLine ( $" FROM " ) ; commandText . AppendLine ( $" \t { queueName } " ) ; commandText . AppendLine ( $" ), TIMEOUT @TimeoutMilliseconds; " ) ; SqlDbFactoryCommand cmd = CreateTextCommand ( commandText . ToString ( ) ) ; if ( _conversationHandle . HasValue ) { cmd . Params . AddGuid ( " " , false , _conversationHandle ) ; } cmd . Params . AddInt64 ( " " , false , ( timeout == - 1 ? timeout : timeout * 1000 ) ) ; cmd . CommandTimeout = ( timeout == - 1 ? 0 : timeout >= int . MaxValue - commandTimeoutExtension ? int . MaxValue : timeout + commandTimeoutExtension ) ; AsyncCallback callBack = new AsyncCallback ( endNotificationListener ) ; _conversationHandle = null ; return cmd . BeginExecuteReader ( callBack , cmd ) ; void endNotificationListener ( IAsyncResult ar ) { try { if DIAGNOSTICS Debug . WriteLine ( $" Listener Reader:  { queueName } " ) ; endif SqlNotificationReceiveEventArgs eventArgs ; using ( cmd ) { try { using ( SqlDataReader reader = cmd . EndExecuteReader ( ar ) ) { if ( reader . Read ( ) ) { eventArgs = new SqlNotificationReceiveEventArgs ( ) ; eventArgs . Status = reader . GetByte ( " " ) ; eventArgs . Priority = reader . GetByte ( " " ) ; eventArgs . QueuingOrder = reader . GetInt64 ( " " ) ; eventArgs . ConversationGroupId = reader . GetGuid ( " " ) ; eventArgs . ConversationHandle = reader . GetGuid ( " " ) ; eventArgs . MessageSequenceNumber = reader . GetInt64 ( " " ) ; eventArgs . ServiceName = reader . GetString ( " " ) ; eventArgs . ServiceId = reader . GetInt32 ( " " ) ; eventArgs . ServiceContractName = reader . GetString ( " " ) ; eventArgs . ServiceContractId = reader . GetInt32 ( " " ) ; eventArgs . MessageTypeName = reader . GetString ( " " ) ; eventArgs . MessageTypeId = reader . GetInt32 ( " " ) ; eventArgs . Validation = reader . GetString ( " " ) ; eventArgs . MessageBody = reader . GetStringNullSafe ( " " ) ; _conversationHandle = eventArgs . ConversationHandle ; if DIAGNOSTICS Debug . WriteLine ( $" Listener Reader: ConversationHandle= { obj . ConversationHandle } , MessageTypeName= { obj . MessageTypeName } , MessageBody= { obj . MessageBody } " ) ; endif } else { eventArgs = null ; } } } catch ( Exception ex ) { throw new DbCommandException ( cmd . Command , ex ) ; } } if ( eventArgs != null ) { NotificationReceived ? . Invoke ( this , eventArgs ) ; } } catch ( Exception ex ) { exceptionCallback ( ex ) ; } } }
Begin execution of a listener for a  that will
execute on a different .
Begin execution of a listener for a that will execute on a different .
// CommandTimeout uses 0 (zero) to define indefinitely. //using (SqlDbFactoryCommand cmd = (SqlDbFactoryCommand)ar.AsyncState) // Send notifications now that the reader has been disposed.
An  that can be used to poll or wait for results, or
both; this value is also needed when invoking ,
which returns a  instance which can be used to retrieve
the returned rows.
An that can be used to poll or wait for results or both ; this value is also needed when invoking which returns a instance which can be used to retrieve the returned rows .
queueName
string
The Service Broker queue name.
The Service Broker queue name .
timeout
int
The number of seconds to wait for the T-SQL RECEIVE statement to
return at least 1 message, or -1 to wait indefinitely.
The number of seconds to wait for the T - SQL RECEIVE statement to return at least 1 message or - 1 to wait indefinitely .
exceptionCallback
Action<Exception>
The callback method for exceptions thrown during the
asynchronous execution period of this call.
The callback method for exceptions thrown during the asynchronous execution period of this call .