1+ #region Usings
2+
3+ using System ;
4+ using System . Text ;
5+ using System . Threading ;
6+ using Extend ;
7+
8+ #endregion
9+
10+ namespace Stomp . Net . Example . SelectorsCore
11+ {
12+ public class Program
13+ {
14+ #region Constants
15+
16+ private const String Host = "host" ;
17+
18+ private const String Password = "password" ;
19+
20+ private const Int32 Port = 63617 ;
21+ //private const Int32 Port = 61613;
22+
23+ private const String QueueName = "TestQ" ;
24+ private const String SelectorKey = "selectorProp" ;
25+ private const String User = "admin" ;
26+
27+ private const Int32 NoOfMessages = 30 ;
28+
29+ private static readonly ManualResetEventSlim _resetEvent = new ManualResetEventSlim ( ) ;
30+
31+ #endregion
32+
33+ public static void Main ( String [ ] args )
34+ {
35+ // Configure a logger to capture the output of the library
36+ Tracer . Trace = new ConsoleLogger ( ) ;
37+
38+ try
39+ {
40+ using ( var subscriber = new Subscriber ( ) )
41+ {
42+ SendMessages ( ) ;
43+
44+ Console . WriteLine ( $ " [{ Thread . CurrentThread . ManagedThreadId } ] Start receiving messages.") ;
45+
46+ subscriber . Start ( ) ;
47+
48+ Console . WriteLine ( _resetEvent . Wait ( 1 . ToMinutes ( ) ) ? "All messages received" : "Timeout :(" ) ;
49+ }
50+ }
51+ catch ( Exception ex )
52+ {
53+ Console . WriteLine ( $ "Error: { ex } ") ;
54+ }
55+
56+ Console . WriteLine ( "Press <enter> to exit." ) ;
57+ Console . ReadLine ( ) ;
58+ }
59+
60+
61+ private static void SendMessages ( )
62+ {
63+ var factory = GetConnectionFactory ( ) ;
64+
65+ // Create connection for both requests and responses
66+ using ( var connection = factory . CreateConnection ( ) )
67+ {
68+ // Open the connection
69+ connection . Start ( ) ;
70+
71+ // Create session for both requests and responses
72+ using ( var session = connection . CreateSession ( AcknowledgementMode . IndividualAcknowledge ) )
73+ {
74+ // Create a message producer
75+ IDestination destinationQueue = session . GetQueue ( QueueName ) ;
76+ using ( var producer = session . CreateProducer ( destinationQueue ) )
77+ {
78+ producer . DeliveryMode = MessageDeliveryMode . Persistent ;
79+
80+ for ( var i = 0 ; i < NoOfMessages ; i ++ )
81+ {
82+ // Send a message to the destination
83+ var message = session . CreateBytesMessage ( Encoding . UTF8 . GetBytes ( $ "Hello World { i } ") ) ;
84+ message . StompTimeToLive = TimeSpan . FromMinutes ( 1 ) ;
85+ message . Headers [ "test" ] = $ "test { i } ";
86+ producer . Send ( message ) ;
87+
88+ Console . WriteLine ( $ "Message sent { i } ") ;
89+ }
90+ }
91+ }
92+ }
93+ }
94+
95+ private static ConnectionFactory GetConnectionFactory ( )
96+ {
97+ // Create a connection factory
98+ var brokerUri = "tcp://" + Host + ":" + Port ;
99+
100+ return new ConnectionFactory ( brokerUri ,
101+ new StompConnectionSettings
102+ {
103+ UserName = User ,
104+ Password = Password ,
105+ TransportSettings =
106+ {
107+ SslSettings =
108+ {
109+ ServerName = "" ,
110+ ClientCertSubject = "" ,
111+ KeyStoreName = "My" ,
112+ KeyStoreLocation = "LocalMachine"
113+ }
114+ } ,
115+ SkipDesinationNameFormatting = false , // Determines whether the destination name formatting should be skipped or not.
116+ SetHostHeader = true , // Determines whether the host header will be added to messages or not
117+ HostHeaderOverride = null // Can be used to override the content of the host header
118+ } ) ;
119+
120+ }
121+
122+ private class Subscriber : IDisposable
123+ {
124+ private IConnection _connection ;
125+ private ISession _session ;
126+ private IMessageConsumer _consumer ;
127+
128+ private Int32 _noOfreceivedMessages ;
129+ private readonly Object _sync = new Object ( ) ;
130+
131+ public void Start ( )
132+ {
133+ var factory = GetConnectionFactory ( ) ;
134+
135+ // Create connection for both requests and responses
136+ _connection = factory . CreateConnection ( ) ;
137+
138+ // Open the connection
139+ _connection . Start ( ) ;
140+
141+ // Create session for both requests and responses
142+ _session = _connection . CreateSession ( AcknowledgementMode . IndividualAcknowledge ) ;
143+
144+ // Create a message consumer
145+ IDestination sourceQueue = _session . GetQueue ( QueueName ) ;
146+ _consumer = _session . CreateConsumer ( sourceQueue ) ;
147+
148+ _consumer . Listener += message =>
149+ {
150+ var content = Encoding . UTF8 . GetString ( message . Content ) ;
151+ Console . WriteLine ( $ " [{ Thread . CurrentThread . ManagedThreadId } ] { content } ") ;
152+
153+ Thread . Sleep ( 500 ) ;
154+
155+ message . Acknowledge ( ) ;
156+
157+ lock ( _sync )
158+ {
159+ _noOfreceivedMessages ++ ;
160+
161+ if ( _noOfreceivedMessages >= NoOfMessages )
162+ _resetEvent . Set ( ) ;
163+ }
164+ } ;
165+ }
166+
167+ #region IDisposable
168+
169+ /// <summary>Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.</summary>
170+ public void Dispose ( )
171+ {
172+ _connection ? . Dispose ( ) ;
173+ _session ? . Dispose ( ) ;
174+ _consumer ? . Dispose ( ) ;
175+ }
176+
177+ #endregion
178+ }
179+ }
180+
181+ /// <summary>
182+ /// Console logger for Stomp.Net
183+ /// </summary>
184+ public class ConsoleLogger : ITrace
185+ {
186+ #region Implementation of ITrace
187+
188+ /// <summary>
189+ /// Writes a message on the error level.
190+ /// </summary>
191+ /// <param name="message">The message.</param>
192+ public void Error ( String message )
193+ => Console . WriteLine ( $ "[Error]\t \t { message } ") ;
194+
195+ /// <summary>
196+ /// Writes a message on the fatal level.
197+ /// </summary>
198+ /// <param name="message">The message.</param>
199+ public void Fatal ( String message )
200+ => Console . WriteLine ( $ "[Fatal]\t \t { message } ") ;
201+
202+ /// <summary>
203+ /// Writes a message on the info level.
204+ /// </summary>
205+ /// <param name="message">The message.</param>
206+ public void Info ( String message )
207+ => Console . WriteLine ( $ "[Info]\t \t { message } ") ;
208+
209+ /// <summary>
210+ /// Writes a message on the warn level.
211+ /// </summary>
212+ /// <param name="message">The message.</param>
213+ public void Warn ( String message )
214+ => Console . WriteLine ( $ "[Warn]\t \t { message } ") ;
215+
216+ #endregion
217+ }
218+ }
0 commit comments