J9sddXLB

1using System;
2using System.Data;
3using System.Data.SqlClient;
4using System.Text;
5
6namespace SqlNativeQL
7{
8    internal class Program
9    {
10        static void Main(string[] args)
11        {
12            var connectionString =
13                @"Data Source=localhost\sqlexpress;Initial Catalog=nservicebus;Integrated Security=True";
14
15            using (var connection = new SqlConnection(connectionString))
16            {
17                connection.Open();
18
19                CreateQueue(connection, "dbo", "sample-queue");
20
21                SendMessage(connection, "dbo.[sample-queue]");
22                SendMessage(connection, "dbo.[sample-queue]");
23
24                var queueLength = GetLength(connection, "dbo.[sample-queue]");
25
26                Console.WriteLine($"Queue length = {queueLength}.");
27            }
28
29            Console.ReadLine();
30        }
31
32        static int GetLength(SqlConnection connection, string queueAddress)
33        {
34            var insertSql = $@"select count(*) from {queueAddress} with (nolock)";
35
36            using (var command = new SqlCommand(insertSql, connection))
37            {
38                return (int) command.ExecuteScalar();
39            }
40        }
41
42        public static void SendMessage(SqlConnection connection, string queueAddress)
43        {
44            var insertSql = $@"
45                    insert into {queueAddress} (
46                        Id,
47                        Recoverable,
48                        Headers,
49                        Body)
50                    values (
51                        @Id,
52                        @Recoverable,
53                        @Headers,
54                        @Body)";
55
56            var headers = @"{ {""NServiceBus.EnclosedMessageTypes"", ""MessageTypeToSend""} }";
57            var body = "{Property:'PropertyValue'}";
58
59            var bytes = Encoding.UTF8.GetBytes(body);
60            using (var command = new SqlCommand(insertSql, connection))
61            {
62                var parameters = command.Parameters;
63                parameters.Add("Id", SqlDbType.UniqueIdentifier).Value = Guid.NewGuid();
64                parameters.Add("Headers", SqlDbType.VarChar).Value = headers;
65                parameters.Add("Body", SqlDbType.VarBinary).Value = bytes;
66                parameters.Add("Recoverable", SqlDbType.Bit).Value = true;
67                command.ExecuteNonQuery();
68            }
69        }
70
71        public static void CreateQueue(SqlConnection connection, string schema, string queueName)
72        {
73            var sql = $@"
74        if not  exists (select * from sys.objects where object_id = object_id(N'[{schema}].[{
75                    queueName
76                }]') and type in (N'U'))
77        begin
78            create table [{schema}].[{queueName}](
79                [Id] [uniqueidentifier] not null,
80                [CorrelationId] [varchar](255),
81                [ReplyToAddress] [varchar](255),
82                [Recoverable] [bit] not null,
83                [Expires] [datetime],
84                [Headers] [nvarchar](max) not null,
85                [Body] [varbinary](max),
86                [RowVersion] [bigint] identity(1,1) not null
87            );
88            create clustered index [Index_RowVersion] on [{schema}].[{queueName}]
89            (
90                [RowVersion]
91            )
92            create nonclustered index [Index_Expires] on [{schema}].[{queueName}]
93            (
94                [Expires]
95            )
96            include
97            (
98                [Id],
99                [RowVersion]
100            )
101            where [Expires] is not null
102        end";
103            using (var command = new SqlCommand(sql, connection))
104            {
105                command.ExecuteNonQuery();
106            }
107        }
108    }
109}