diff options
| author | Ted Ross <tross@apache.org> | 2010-05-28 18:09:10 +0000 |
|---|---|---|
| committer | Ted Ross <tross@apache.org> | 2010-05-28 18:09:10 +0000 |
| commit | 947b1491d7a39c87c4560126a6e50646aa2a2b24 (patch) | |
| tree | 19a5c7ac347d807125f8352ef5f2d41810d79281 /cpp/bindings/qpid/dotnet/examples/csharp.map.callback.receiver/csharp.map.callback.receiver.cs | |
| parent | 8d317104053b8258380c47af8d792517c4da10b7 (diff) | |
| download | qpid-python-947b1491d7a39c87c4560126a6e50646aa2a2b24.tar.gz | |
QPID-2628 - Patch from Chuck Rolke
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@949245 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'cpp/bindings/qpid/dotnet/examples/csharp.map.callback.receiver/csharp.map.callback.receiver.cs')
| -rw-r--r-- | cpp/bindings/qpid/dotnet/examples/csharp.map.callback.receiver/csharp.map.callback.receiver.cs | 280 |
1 files changed, 280 insertions, 0 deletions
diff --git a/cpp/bindings/qpid/dotnet/examples/csharp.map.callback.receiver/csharp.map.callback.receiver.cs b/cpp/bindings/qpid/dotnet/examples/csharp.map.callback.receiver/csharp.map.callback.receiver.cs new file mode 100644 index 0000000000..e7294c6e1a --- /dev/null +++ b/cpp/bindings/qpid/dotnet/examples/csharp.map.callback.receiver/csharp.map.callback.receiver.cs @@ -0,0 +1,280 @@ +/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+using System;
+using System.Collections.Generic;
+using org.apache.qpid.messaging;
+using org.apache.qpid.messaging.sessionreceiver;
+
+namespace org.apache.qpid.messaging.examples
+{
+ /// <summary>
+ /// A class with functions to display structured messages.
+ /// </summary>
+ public static class MessageViewer
+ {
+ /// <summary>
+ /// A Function to display a amqp/map message packaged as a Dictionary.
+ /// </summary>
+ /// <param name="dict">The AMQP map</param>
+ /// <param name="level">Nested depth</param>
+ public static void ShowDictionary(Dictionary<string, object> dict, int level)
+ {
+ foreach (KeyValuePair<string, object> kvp in dict)
+ {
+ Console.Write(new string(' ', level * 4));
+
+ if (QpidTypeCheck.ObjectIsMap(kvp.Value))
+ {
+ Console.WriteLine("Key: {0}, Value: Dictionary", kvp.Key);
+ ShowDictionary((Dictionary<string, object>)kvp.Value, level + 1);
+ }
+ else if (QpidTypeCheck.ObjectIsList(kvp.Value))
+ {
+ Console.WriteLine("Key: {0}, Value: List", kvp.Key);
+ ShowList((List<object>)kvp.Value, level + 1);
+ }
+ else
+ Console.WriteLine("Key: {0}, Value: {1}, Type: {2}",
+ kvp.Key, kvp.Value, kvp.Value.GetType().ToString());
+ }
+ }
+
+ /// <summary>
+ /// A function to display a ampq/list message packaged as a List.
+ /// </summary>
+ /// <param name="list">The AMQP list</param>
+ /// <param name="level">Nested depth</param>
+ public static void ShowList(List<object> list, int level)
+ {
+ foreach (object obj in list)
+ {
+ Console.Write(new string(' ', level * 4));
+
+ if (QpidTypeCheck.ObjectIsMap(obj))
+ {
+ Console.WriteLine("Dictionary");
+ ShowDictionary((Dictionary<string, object>)obj, level + 1);
+ }
+ else if (QpidTypeCheck.ObjectIsList(obj))
+ {
+ Console.WriteLine("List");
+ ShowList((List<object>)obj, level + 1);
+ }
+ else
+ Console.WriteLine("Value: {0}, Type: {1}",
+ obj.ToString(), obj.GetType().ToString());
+ }
+ }
+
+ /// <summary>
+ /// A function to diplay a Message. The native Object type is
+ /// decomposed into AMQP types.
+ /// </summary>
+ /// <param name="message">The Message</param>
+ public static void ShowMessage(Message message)
+ {
+ if ("amqp/map" == message.getContentType())
+ {
+ Console.WriteLine("Received a Dictionary");
+ Dictionary<string, object> content = new Dictionary<string, object>();
+ message.getContent(content);
+ ShowDictionary(content, 0);
+ }
+ else if ("amqp/list" == message.getContentType())
+ {
+ Console.WriteLine("Received a List");
+ List<object> content = new List<object>();
+ message.getContent(content);
+ ShowList(content, 0);
+ }
+ else
+ {
+ Console.WriteLine("Received a String");
+ Console.WriteLine(message.getContent());
+ }
+ }
+ }
+
+
+
+ /// <summary>
+ /// A model class to demonstrate how a user may use the Qpid Messaging
+ /// interface to receive Session messages using a callback.
+ /// </summary>
+ class ReceiverProcess : ISessionReceiver
+ {
+ UInt32 messagesReceived = 0;
+
+ /// <summary>
+ /// SessionReceiver implements the ISessionReceiver interface.
+ /// It is the callback function that receives all messages for a Session.
+ /// It may be called any time server is running.
+ /// It is always called on server's private thread.
+ /// </summary>
+ /// <param name="receiver">The Receiver associated with the message.</param>
+ /// <param name="message">The Message</param>
+ public void SessionReceiver(Receiver receiver, Message message)
+ {
+ //
+ // Indicate message reception
+ //
+ Console.WriteLine("--- Message {0}", ++messagesReceived);
+
+ //
+ // Display the received message
+ //
+ MessageViewer.ShowMessage(message);
+
+ //
+ // Acknowledge the receipt of all received messages.
+ //
+ receiver.getSession().acknowledge();
+ }
+
+
+ /// <summary>
+ /// Usage
+ /// </summary>
+ /// <param name="url">Connection target</param>
+ /// <param name="addr">Address: broker exchange + routing key</param>
+ /// <param name="nSec">n seconds to keep callback open</param>
+ static void usage(string url, string addr, int nSec)
+ {
+
+ Console.WriteLine("usage: {0} [url [addr [nSec]]]",
+ System.Diagnostics.Process.GetCurrentProcess().ProcessName);
+ Console.WriteLine();
+ Console.WriteLine("A program to connect to a broker and receive");
+ Console.WriteLine("messages from a named exchange with a routing key.");
+ Console.WriteLine("The receiver uses a session callback and keeps the callback");
+ Console.WriteLine("server open for so many seconds.");
+ Console.WriteLine("The details of the message body's types and values are shown.");
+ Console.WriteLine();
+ Console.WriteLine(" url = target address for 'new Connection(url)'");
+ Console.WriteLine(" addr = address for 'session.createReceiver(addr)'");
+ Console.WriteLine(" nSec = time in seconds to keep the receiver callback open");
+ Console.WriteLine();
+ Console.WriteLine("Default values:");
+ Console.WriteLine(" {0} {1} {2} {3}",
+ System.Diagnostics.Process.GetCurrentProcess().ProcessName,
+ url, addr, nSec);
+ }
+
+
+ /// <summary>
+ /// A function to illustrate how to open a Session callback and
+ /// receive messages.
+ /// </summary>
+ /// <param name="args">Main program arguments</param>
+ public void TestProgram(string[] args)
+ {
+ string url = "amqp:tcp:localhost:5672";
+ string addr = "amq.direct/map_example";
+ int nSec = 30;
+
+ if (1 == args.Length)
+ {
+ if (args[0].Equals("-h") || args[0].Equals("-H") || args[0].Equals("/?"))
+ {
+ usage(url, addr, nSec);
+ return;
+ }
+ }
+
+ if (args.Length > 0)
+ url = args[0];
+ if (args.Length > 1)
+ addr = args[1];
+ if (args.Length > 2)
+ nSec = System.Convert.ToInt32(args[2]);
+
+ //
+ // Create and open an AMQP connection to the broker URL
+ //
+ Connection connection = new Connection(url);
+ connection.open();
+
+ //
+ // Create a session.
+ //
+ Session session = connection.createSession();
+
+ //
+ // Receive through callback
+ //
+ // Create callback server and implicitly start it
+ //
+ sessionreceiver.server cbServer =
+ new sessionreceiver.server(session, this);
+
+ //
+ // The callback server is running and executing callbacks on a
+ // separate thread.
+ //
+
+ //
+ // Create a receiver for the direct exchange using the
+ // routing key "map_example".
+ //
+ Receiver receiver = session.createReceiver(addr);
+
+ //
+ // Establish a capacity
+ //
+ receiver.setCapacity(100);
+
+ //
+ // Wait so many seconds for messages to arrive.
+ //
+ System.Threading.Thread.Sleep(nSec * 1000); // in mS
+
+ //
+ // Stop the callback server.
+ //
+ cbServer.close();
+
+ //
+ // Close the receiver and the connection.
+ //
+ receiver.close();
+ connection.close();
+ }
+ }
+
+
+ class MapCallbackReceiverMain
+ {
+ /// <summary>
+ /// Main program
+ /// </summary>
+ /// <param name="args">Main prgram args</param>
+ static void Main(string[] args)
+ {
+ // Invoke 'TestProgram' as non-static class.
+ ReceiverProcess mainProc = new ReceiverProcess();
+
+ mainProc.TestProgram(args);
+
+ }
+ }
+}
+
|