blob: 705e7a9aab7d7e22db40016935c6985d4a321215 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
|
using System;
using System.IO;
using System.Runtime.Serialization.Formatters.Binary;
using System.Transactions;
using ProtoBuf;
using Rhino.Queues;
namespace common
{
public class RhinoPublisher : ServiceBus
{
BinaryFormatter formatter = new BinaryFormatter();
readonly int port;
string destination_queue;
IQueueManager sender;
public RhinoPublisher(string destination_queue, int port, IQueueManager manager)
{
this.port = port;
this.destination_queue = destination_queue;
sender = manager;
}
public void publish<T>() where T : new()
{
publish(new T());
}
public void publish<T>(T item) where T : new()
{
using (var transaction = new TransactionScope())
{
var destination = "rhino.queues://localhost:{0}/{1}".format(port, destination_queue);
this.log().debug("sending {0} to {1}", item, destination);
sender.Send(new Uri(destination), create_payload_from(item));
transaction.Complete();
}
}
MessagePayload create_payload_from<T>(T item)
{
using (var stream = new MemoryStream())
{
Serializer.Serialize(stream, item);
//formatter.Serialize(stream, item);
var payload = new MessagePayload {Data = stream.ToArray()};
payload.Headers["type"] = typeof (T).FullName;
return payload;
}
}
public void publish<T>(Action<T> configure) where T : new()
{
var item = new T();
configure(item);
publish(item);
}
}
}
|
