WCF Extensibility – Transport Channels – Duplex Channels

Article
01/10/2012

This post is part of a series about WCF extensibility points. For a list of all previous posts and planned future ones, go to the index page .

And we’ll get to the end of the channel layer and transports in general with this post. After looking in many details at two types of transport channel, I’ll close this section of the series with a look at a duplex channel. Duplex channels are those in which, once the connection is established between the client and the server, the latter doesn’t need to wait for messages from the client to send something. Instead, the server can send many messages to the client without being requested.

A duplex channel is essentially a pair of channels, one for each side of the communication, which can be used independently – the interface IDuplexChannel is indeed defined without any members of its own, simply inheriting from IInputChannel and IOutputChannel (and also IChannel and ICommunicationObject, but both input and output channel interfaces already inherit from them). It means that they can be used in both at the client and the server for contracts which use those channel shapes, such as one-way operations (for a small recap on channel shapes, see the first post about channels).

But duplex channels can be used for more “normal” cases as well, such as calling request/reply operations. The WCF channel has a multiplexer which will add a message id to outgoing messages and then correlate that message with incoming ones to be able to “respond” to a request. Notice that in order for this to work, the messages must support some form addressing (i.e., the addressing version of the message version of the encoder used by the transport must not be None).

This post will describe both input and output channel interfaces, and on the sample section I’ll show a simple duplex transport channel (based on the same framing used in the other transport channel posts) to illustrate its usage.

Public implementations in WCF

None. As with most channel extensibility points, there are no public implementations in WCF. There are many internal implementations, such as the one used by the TCP and named pipes transport.

Interface definition

public interface IDuplexChannel : IInputChannel, IOutputChannel, IChannel, ICommunicationObject
{
}

The IDuplexChannel interface is literally empty as I mentioned above, so there’s nothing really to talk about it. It’s more interesting to talk about its components, so there we go.

public interface IOutputChannel : IChannel, ICommunicationObject
{
// Methods
IAsyncResult BeginSend(Message message, AsyncCallback callback, object state);
IAsyncResult BeginSend(Message message, TimeSpan timeout, AsyncCallback callback, object state);
void EndSend(IAsyncResult result);
void Send(Message message);
void Send(Message message, TimeSpan timeout);
// Properties
EndpointAddress RemoteAddress { get; }
Uri Via { get; }
}

The IOutputChannel interface is quite similar to the IRequestChannel interface, covered in the first post about transport channels. The main difference is that the output channel only sends messages, but doesn’t receive any responses to it. The properties are similar: RemoteAddress returns the address to which the channel is sending messages to, while Via is the transport address to which the request is sent. They are usually the same, but in cases where a message goes through multiple hops before reaching its final destination (i.e., in a proxying or routing scenario), the Via property points to the “next step”, while RemoteAddress points to the final destination of the message.

The output channel has also essentially one method, used to send messages, in various flavors. It can be synchronous (Send) or asynchronous (BeginSend / EndSend), and it can take a timeout parameter or have the channel assume a default timeout.

public interface IInputChannel : IChannel, ICommunicationObject
{
// Methods
IAsyncResult BeginReceive(AsyncCallback callback, object state);
IAsyncResult BeginReceive(TimeSpan timeout, AsyncCallback callback, object state);
IAsyncResult BeginTryReceive(TimeSpan timeout, AsyncCallback callback, object state);
IAsyncResult BeginWaitForMessage(TimeSpan timeout, AsyncCallback callback, object state);
Message EndReceive(IAsyncResult result);
bool EndTryReceive(IAsyncResult result, out Message message);
bool EndWaitForMessage(IAsyncResult result);
Message Receive();
Message Receive(TimeSpan timeout);
bool TryReceive(TimeSpan timeout, out Message message);
bool WaitForMessage(TimeSpan timeout);
// Properties
EndpointAddress LocalAddress { get; }
}

The IInputChannel interface, on the other hand, is similar to the IReplyChannel interface, covered in the post about reply transport channels. Again, the main difference is that it just receives a message, without sending any response, so the *Receive methods don’t return any request context objects like in the reply channels. Like on the reply channel, there are both synchronous and asynchronous versions of the simple “give me a message” (Receive and BeginReceive / EndReceive, respectively) which throw an exception if a message isn’t received after a certain timeout. Also, like the reply channels, input channels also have the exception-less variations of TryReceive (and the asynchronous equivalent BeginTryReceive / EndTryReceive), which return a message if it arrives within the specified timeout, and the WaitForMessage (and its asynchronous counterparts BeginWaitForMessage / EndWaitForMessage) which peek on the input queue and tells the caller whether a message is available, but doesn’t return it (the caller needs to call another of the input channel methods to retrieve the message).

How to add a duplex transport channel

It’s the same as the other channels: you transport channels are added by using a channel factory (client) / or channel listener (server), which are created by a custom binding element. The binding element corresponding to a transport channel must be derived from the TransportBindingElement class. Besides the normal properties from the base BindingElement class, the transport binding element also defines a Scheme property, which needs to return the URI scheme for the transport. The sample code has an example of defining and adding a custom duplex transport channel. On the client side the binding element class must override both CanBuildChannelFactory<TChannel> (to tell the caller that it supports the IDuplexChannel interface) and BuildChannelFactory<TChannel> to actually build the factory which can build the channels. On the server side the binding element must do the listener counterparts, overriding CanBuildChannelListener<TChannel> (to tell the caller that it supports creating duplex channels) and BuildChannelListener<TChannel> to actually build the listener for such channels.

Real world scenario: implementing a custom transport protocol

I find it hard calling it a real world scenario, since I haven’t found any requests for this in forums or otherwise, so let’s treat it as a sample instead. Similar to the JSON-RPC scenario, we’ll create a custom transport protocol (the same, actually, with a 4-byte length, followed by the actual message). This time, however, we won’t be able to use the ByteStreamMessageEncodingBindingElement which we used in the JSON-RPC case, because, as I mentioned before, we need some addressing (which WCF will use to correlate between requests and replies). So for this scenario we’ll use a new encoding – the default will now be the MessageVersion.Default, which is the same as MessageVersion.Soap12WSAddressing10. The advantage of it is that, by using a format which is supported out-of-the-box, we don’t need any of the service model extensions which we did on the previous sample.

Even though the goal of having a duplex transport is to be able to do full duplex communication, I’ll actually start with a simple (e.g., request / reply) contract, since it’s a lot easier to test those by themselves (using a socket-based client or server, like the one created for the JSON-RPC sample). So we’ll start with that (that’s how I created the sample, so it makes sense for me to write about it this way).

And before we dive into the code, the usual disclaimer: this is a sample for illustrating the topic of this post, this is not production-ready code. I tested it for a few contracts and it worked, but I cannot guarantee that it will work for all scenarios (please let me know if you find a bug or something missing). I really kept the error checking to a minimum (especially in the “simple service” project and in many other helper functions whose inputs should be sanitized, to make the sample small. Also, I practically didn’t implement any timeouts in the channel (they’re mostly being ignored), otherwise this sample which is rather large would be even more complex.

Let’s start with a simple contract. To prevent service-model issues from validating it, let’s start with an untyped message contract.

[ServiceContract]
public interface IUntypedTest
{
[OperationContract(Action = "*", ReplyAction = "*")]
Message Process(Message input);
}
public class Service : IUntypedTest
{
public Message Process(Message input)
{
Message result = Message.CreateMessage(input.Version, "ReplyAction", "The response");
result.Headers.To = input.Headers.ReplyTo.Uri;
return result;
}
}

With that contract, we can try to create a client which uses that new transport and consume that service.

static void TestWithUntypedMessage()
{
string baseAddress = SizedTcpDuplexTransportBindingElement.SizedTcpScheme + "://localhost:8000";
ServiceHost host = new ServiceHost(typeof(Service), new Uri(baseAddress));
MessageEncodingBindingElement encodingBE = new TextMessageEncodingBindingElement();
Binding binding = new CustomBinding(encodingBE, new SizedTcpDuplexTransportBindingElement());
host.AddServiceEndpoint(typeof(IUntypedTest), binding, "");
host.Open();
Console.WriteLine("Host opened");
Socket socket = GetConnectedSocket(8000);
Message input = Message.CreateMessage(MessageVersion.Soap12WSAddressing10, "myAction", "Hello world");
input.Headers.To = new Uri(baseAddress);
input.Headers.ReplyTo = new EndpointAddress("https://www.w3.org/2005/08/addressing/anonymous");
MessageEncoder encoder = encodingBE.CreateMessageEncoderFactory().Encoder;
BufferManager bufferManager = BufferManager.CreateBufferManager(int.MaxValue, int.MaxValue);
ArraySegment<byte> encoded = encoder.WriteMessage(input, int.MaxValue, bufferManager, 4);
Formatting.SizeToBytes(encoded.Count, encoded.Array, 0);
Console.WriteLine("Sending those bytes:");
Debugging.PrintBytes(encoded.Array, encoded.Count + encoded.Offset);
socket.Send(encoded.Array, 0, encoded.Offset + encoded.Count, SocketFlags.None);
byte[] recvBuffer = new byte[10000];
int bytesRecvd = socket.Receive(recvBuffer);
Console.WriteLine("Received {0} bytes", bytesRecvd);
Debugging.PrintBytes(recvBuffer, bytesRecvd);
socket.Close();
Console.WriteLine("Press ENTER to close");
Console.ReadLine();
host.Close();
}

Ok, that is the goal of the first “milestone”. Let’s start with the binding element class. In the example above we only need to override the channel listener part of the binding element (since we’re only using it in the server side), telling the caller that we do support a duplex channel (at CanBuildChannelListener<TChannel>) and create it when requested (at BuildChannelListener<TChannel>).

public class SizedTcpDuplexTransportBindingElement : TransportBindingElement
{
public const string SizedTcpScheme = "sized.tcp";
public override string Scheme
{
get { return SizedTcpScheme; }
}
public override bool CanBuildChannelListener<TChannel>(BindingContext context)
{
return typeof(TChannel) == typeof(IDuplexChannel);
}
public override IChannelListener<TChannel> BuildChannelListener<TChannel>(BindingContext context)
{
return (IChannelListener<TChannel>)(object)new SizedTcpDuplexChannelListener(this, context);
}
}

The listener class is similar to the listener class used in the example for the reply channel. The main difference here is that instead of enforcing the byte stream message encoder, this channel can work with any encoding as long as its message version is the expected one. That means that we’ll be able to replace the TextMessageEncodingBindingElement with a BinaryMessageEncodingBindingElement, for example, and the code should work just fine.

class SizedTcpDuplexChannelListener : ChannelListenerBase<IDuplexChannel>
{
BufferManager bufferManager;
MessageEncoderFactory encoderFactory;
Socket listenSocket;
Uri uri;
public SizedTcpDuplexChannelListener(SizedTcpDuplexTransportBindingElement bindingElement, BindingContext context)
: base(context.Binding)
{
// populate members from binding element
int maxBufferSize = (int)bindingElement.MaxReceivedMessageSize;
this.bufferManager = BufferManager.CreateBufferManager(bindingElement.MaxBufferPoolSize, maxBufferSize);
Collection<MessageEncodingBindingElement> messageEncoderBindingElements
= context.BindingParameters.FindAll<MessageEncodingBindingElement>();
if (messageEncoderBindingElements.Count > 1)
{
throw new InvalidOperationException("More than one MessageEncodingBindingElement was found in the BindingParameters of the BindingContext");
}
else if (messageEncoderBindingElements.Count == 1)
{
if (messageEncoderBindingElements[0].MessageVersion != MessageVersion.Soap12WSAddressing10)
{
throw new InvalidOperationException("This transport must be used with the an encoding with MessageVersion.Soap12WSAddressing10.");
}
this.encoderFactory = messageEncoderBindingElements[0].CreateMessageEncoderFactory();
}
else
{
this.encoderFactory = new TextMessageEncodingBindingElement(MessageVersion.Soap12WSAddressing10, Encoding.UTF8).CreateMessageEncoderFactory();
}
this.uri = new Uri(context.ListenUriBaseAddress, context.ListenUriRelativeAddress);
}
protected override IDuplexChannel OnAcceptChannel(TimeSpan timeout)
{
try
{
Socket dataSocket = listenSocket.Accept();
return new SizedTcpDuplexServerChannel(this.encoderFactory.Encoder, this.bufferManager, new EndpointAddress(this.uri), dataSocket, this);
}
catch (ObjectDisposedException)
{
// socket closed
return null;
}
}
protected override IAsyncResult OnBeginAcceptChannel(TimeSpan timeout, AsyncCallback callback, object state)
{
return new AcceptChannelAsyncResult(timeout, this, callback, state);
}
protected override IDuplexChannel OnEndAcceptChannel(IAsyncResult result)
{
return AcceptChannelAsyncResult.End(result);
}
}

The channel created on OnAcceptChannel is of type SizedTcpDuplexServerChannel, which is listed below. It will extend the “base” duplex channel (to be shown later), again, similar to the reply channel sample, to fill up the To header and the RemoteEndpointMessageProperty on incoming messages.

class SizedTcpDuplexServerChannel : SizedTcpDuplexChannel
{
Socket dataSocket;
public SizedTcpDuplexServerChannel(MessageEncoder messageEncoder, BufferManager bufferManager, EndpointAddress localAddress, Socket dataSocket, ChannelManagerBase channelManager)
:base(messageEncoder, bufferManager, channelManager, localAddress, null, null)
{
this.dataSocket = dataSocket;
this.InitializeSocket(dataSocket);
}
protected override Message DecodeMessage(ArraySegment<byte> data)
{
Message result = base.DecodeMessage(data);
if (result != null)
{
result.Headers.To = this.LocalAddress.Uri;
IPEndPoint remoteEndpoint = (IPEndPoint)this.dataSocket.RemoteEndPoint;
RemoteEndpointMessageProperty property = new RemoteEndpointMessageProperty(remoteEndpoint.Address.ToString(), remoteEndpoint.Port);
result.Properties.Add(RemoteEndpointMessageProperty.Name, property);
}
return result;
}
}

The main duplex channel class inherits from the same SizedTcpBaseChannel which was first introduced in the post about request channels. Notice that the implementation of the duplex channel implementation is quite simple – since all the main logic has already been written on the base class.

abstract class SizedTcpDuplexChannel : SizedTcpBaseChannel, IDuplexChannel
{
Socket socket;
Uri via;
EndpointAddress localAddress;
EndpointAddress remoteAddress;
public SizedTcpDuplexChannel(
MessageEncoder encoder,
BufferManager bufferManager,
ChannelManagerBase channelManager,
EndpointAddress localAddress,
EndpointAddress remoteAddress,
Uri via)
: base(encoder, bufferManager, channelManager)
{
this.via = via;
this.remoteAddress = remoteAddress;
this.localAddress = localAddress;
}
protected override void InitializeSocket(Socket socket)
{
this.socket = socket;
base.InitializeSocket(socket);
if (this.remoteAddress == null)
{
IPEndPoint remoteEndpoint = (IPEndPoint)socket.RemoteEndPoint;
UriBuilder builder = new UriBuilder(
SizedTcpDuplexTransportBindingElement.SizedTcpScheme,
remoteEndpoint.Address.ToString(),
remoteEndpoint.Port);
this.remoteAddress = new EndpointAddress(builder.Uri);
}
if (this.localAddress == null)
{
IPEndPoint localEndpoint = (IPEndPoint)socket.LocalEndPoint;
UriBuilder builder = new UriBuilder(
SizedTcpDuplexTransportBindingElement.SizedTcpScheme,
localEndpoint.Address.ToString(),
localEndpoint.Port);
this.localAddress = new EndpointAddress(builder.Uri);
}
}
#region IInputChannel Members
public IAsyncResult BeginReceive(TimeSpan timeout, AsyncCallback callback, object state)
{
return base.BeginReceiveMessage(timeout, callback, state);
}
public IAsyncResult BeginReceive(AsyncCallback callback, object state)
{
return this.BeginReceive(this.DefaultReceiveTimeout, callback, state);
}
public IAsyncResult BeginTryReceive(TimeSpan timeout, AsyncCallback callback, object state)
{
return new TryReceiveAsyncResult(timeout, this, callback, state);
}
public IAsyncResult BeginWaitForMessage(TimeSpan timeout, AsyncCallback callback, object state)
{
throw new NotSupportedException("No peeking support");
}
public Message EndReceive(IAsyncResult result)
{
return base.EndReceiveMessage(result);
}
public bool EndTryReceive(IAsyncResult result, out Message message)
{
return TryReceiveAsyncResult.End(result, out message);
}
public bool EndWaitForMessage(IAsyncResult result)
{
throw new NotSupportedException("No peeking support");
}
public EndpointAddress LocalAddress
{
get { return this.localAddress; }
}
public Message Receive(TimeSpan timeout)
{
return base.ReceiveMessage(timeout);
}
public Message Receive()
{
return this.Receive(this.DefaultReceiveTimeout);
}
public bool TryReceive(TimeSpan timeout, out Message message)
{
try
{
message = base.ReceiveMessage(timeout);
return true;
}
catch (TimeoutException)
{
message = null;
return false;
}
}
public bool WaitForMessage(TimeSpan timeout)
{
throw new NotSupportedException("No peeking support");
}
#endregion
#region IOutputChannel Members
public IAsyncResult BeginSend(Message message, TimeSpan timeout, AsyncCallback callback, object state)
{
return base.BeginSendMessage(message, timeout, callback, state);
}
public IAsyncResult BeginSend(Message message, AsyncCallback callback, object state)
{
return this.BeginSend(message, this.DefaultSendTimeout, callback, state);
}
public void EndSend(IAsyncResult result)
{
base.EndSendMessage(result);
}
public EndpointAddress RemoteAddress
{
get { return this.remoteAddress; }
}
public void Send(Message message, TimeSpan timeout)
{
base.SendMessage(message, timeout);
}
public void Send(Message message)
{
this.Send(message, this.DefaultSendTimeout);
}
public Uri Via
{
get { return this.via; }
}
#endregion
}

Time to try the first program. If everything was coded correctly, we should see a response from the service.

Received 410 bytes
00 00 01 96 3C 73 3A 45 6E 76 65 6C 6F 70 65 20 ....<s:E nvelope
78 6D 6C 6E 73 3A 73 3D 22 68 74 74 70 3A 2F 2F xmlns:s= "https://
77 77 77 2E 77 33 2E 6F 72 67 2F 32 30 30 33 2F www.w3.o rg/2003/
30 35 2F 73 6F 61 70 2D 65 6E 76 65 6C 6F 70 65 05/soap- envelope
22 20 78 6D 6C 6E 73 3A 61 3D 22 68 74 74 70 3A " xmlns: a="http:
2F 2F 77 77 77 2E 77 33 2E 6F 72 67 2F 32 30 30 //www.w3 .org/200
35 2F 30 38 2F 61 64 64 72 65 73 73 69 6E 67 22 5/08/add ressing"
3E 3C 73 3A 48 65 61 64 65 72 3E 3C 61 3A 41 63 ><s:Head er><a:Ac
74 69 6F 6E 20 73 3A 6D 75 73 74 55 6E 64 65 72 tion s:m ustUnder
73 74 61 6E 64 3D 22 31 22 3E 52 65 70 6C 79 41 stand="1 ">ReplyA
63 74 69 6F 6E 3C 2F 61 3A 41 63 74 69 6F 6E 3E ction</a :Action>
3C 61 3A 54 6F 20 73 3A 6D 75 73 74 55 6E 64 65 <a:To s: mustUnde
72 73 74 61 6E 64 3D 22 31 22 3E 68 74 74 70 3A rstand=" 1">http:
2F 2F 73 63 68 65 6D 61 73 2E 6D 69 63 72 6F 73 //schema s.micros
6F 66 74 2E 63 6F 6D 2F 32 30 30 35 2F 31 32 2F oft.com/ 2005/12/
53 65 72 76 69 63 65 4D 6F 64 65 6C 2F 41 64 64 ServiceM odel/Add
72 65 73 73 69 6E 67 2F 41 6E 6F 6E 79 6D 6F 75 ressing/ Anonymou
73 3C 2F 61 3A 54 6F 3E 3C 2F 73 3A 48 65 61 64 s</a:To> </s:Head
65 72 3E 3C 73 3A 42 6F 64 79 3E 3C 73 74 72 69 er><s:Bo dy><stri
6E 67 20 78 6D 6C 6E 73 3D 22 68 74 74 70 3A 2F ng xmlns ="http:/
2F 73 63 68 65 6D 61 73 2E 6D 69 63 72 6F 73 6F /schemas .microso
66 74 2E 63 6F 6D 2F 32 30 30 33 2F 31 30 2F 53 ft.com/2 003/10/S
65 72 69 61 6C 69 7A 61 74 69 6F 6E 2F 22 3E 54 erializa tion/">T
68 65 20 72 65 73 70 6F 6E 73 65 3C 2F 73 74 72 he respo nse</str
69 6E 67 3E 3C 2F 73 3A 42 6F 64 79 3E 3C 2F 73 ing></s: Body></s
3A 45 6E 76 65 6C 6F 70 65 3E :Envelop e>

One thing I pointed before was that since we’re using the SOAP message version, we wouldn’t need to use any service model extensibility and it should just work. Time to try it.

[ServiceContract]
public interface ITypedTest
{
[OperationContract]
int Add(int x, int y);
[OperationContract]
int Subtract(int x, int y);
}
public class Service : ITypedTest
{
public int Add(int x, int y)
{
return x + y;
}
public int Subtract(int x, int y)
{
return x - y;
}
}

Now we can create a client to test it. We need to create a request in the format expected by the server. One easy way to do that is to create a server using a binding such as WSHttpBinding (which uses the same message version), capture a request sent to the same contract, then replicate it. That’s what was done in the code below.

static void TestWithTypedMessage()
{
string baseAddress = SizedTcpDuplexTransportBindingElement.SizedTcpScheme + "://localhost:8000";
ServiceHost host = new ServiceHost(typeof(Service), new Uri(baseAddress));
Binding binding = new CustomBinding(new SizedTcpDuplexTransportBindingElement());
host.AddServiceEndpoint(typeof(ITypedTest), binding, "");
host.Open();
Console.WriteLine("Host opened");
Socket socket = GetConnectedSocket(8000);
string request = @"<s:Envelope
xmlns:s=""https://www.w3.org/2003/05/soap-envelope""
xmlns:a=""https://www.w3.org/2005/08/addressing"">
<s:Header>
<a:Action s:mustUnderstand=""1"">https://tempuri.org/ITypedTest/Add</a:Action>
<a:MessageID>urn:uuid:c2998797-7312-481a-8f73-230406b12bea</a:MessageID>
<a:ReplyTo>
<a:Address>https://www.w3.org/2005/08/addressing/anonymous</a:Address>
</a:ReplyTo>
<a:To s:mustUnderstand=""1"">ENDPOINT_ADDRESS</a:To>
</s:Header>
<s:Body>
<Add xmlns=""https://tempuri.org/"">
<x>4</x>
<y>5</y>
</Add>
</s:Body>
</s:Envelope>";
request = request.Replace("ENDPOINT_ADDRESS", baseAddress);
Encoding encoding = new UTF8Encoding(false);
int byteCount = encoding.GetByteCount(request);
byte[] reqBytes = new byte[byteCount + 4];
Formatting.SizeToBytes(byteCount, reqBytes, 0);
encoding.GetBytes(request, 0, request.Length, reqBytes, 4);
Console.WriteLine("Sending those bytes:");
Debugging.PrintBytes(reqBytes);
socket.Send(reqBytes);
byte[] recvBuffer = new byte[10000];
int bytesRecvd = socket.Receive(recvBuffer);
Console.WriteLine("Received {0} bytes", bytesRecvd);
Debugging.PrintBytes(recvBuffer, bytesRecvd);
Console.WriteLine("Press ENTER to send another request");
Console.ReadLine();
request = @"<s:Envelope
xmlns:s=""https://www.w3.org/2003/05/soap-envelope""
xmlns:a=""https://www.w3.org/2005/08/addressing"">
<s:Header>
<a:Action s:mustUnderstand=""1"">https://tempuri.org/ITypedTest/Subtract</a:Action>
<a:MessageID>urn:uuid:c2998797-7312-481a-8f73-230406b12bea</a:MessageID>
<a:ReplyTo>
<a:Address>https://www.w3.org/2005/08/addressing/anonymous</a:Address>
</a:ReplyTo>
<a:To s:mustUnderstand=""1"">ENDPOINT_ADDRESS</a:To>
</s:Header>
<s:Body>
<Subtract xmlns=""https://tempuri.org/"">
<x>4</x>
<y>5</y>
</Subtract>
</s:Body>
</s:Envelope>";
request = request.Replace("ENDPOINT_ADDRESS", baseAddress);
byteCount = encoding.GetByteCount(request);
reqBytes = new byte[byteCount + 4];
Formatting.SizeToBytes(byteCount, reqBytes, 0);
encoding.GetBytes(request, 0, request.Length, reqBytes, 4);
Console.WriteLine("Sending those bytes:");
Debugging.PrintBytes(reqBytes);
socket.Send(reqBytes);
bytesRecvd = socket.Receive(recvBuffer);
Console.WriteLine("Received {0} bytes", bytesRecvd);
Debugging.PrintBytes(recvBuffer, bytesRecvd);
socket.Close();
Console.WriteLine("Press ENTER to close");
Console.ReadLine();
host.Close();
}

Ok, now we know that the server side is working. Next time, move to the client. To test that, I used a similar “simple” socket server as the sample for request channels.

[ServiceContract]
public interface ITypedTest
{
[OperationContract]
int Add(int x, int y);
[OperationContract]
int Subtract(int x, int y);
}
class Program
{
static void Main(string[] args)
{
SocketsServer s = new SocketsServer(8000);
s.StartServing();
Binding binding = new CustomBinding(new SizedTcpDuplexTransportBindingElement());
string address = SizedTcpDuplexTransportBindingElement.SizedTcpScheme + "://localhost:8000";
ChannelFactory<ITypedTest> factory = new ChannelFactory<ITypedTest>(binding, new EndpointAddress(address));
ITypedTest proxy = factory.CreateChannel();
Console.WriteLine(proxy.Add(4, 5));
Console.WriteLine(proxy.Subtract(44, 66));
s.StopServing();
}
}

Trying to run the code above, it didn’t work – with the following error: “System.InvalidOperationException: The binding (Name=CustomBinding, Namespace=https://tempuri.org/) cannot be used to create a ChannelFactory or a ChannelListener because it appears to be missing a TransportBindingElement. Every binding must have at least one binding element that derives from TransportBindingElement”. The error message isn’t great, because we do have a transport binding element in the binding (it’s the only one). The problem is that it doesn’t create any channel factory, since we didn’t configure it to do so. With the two overrides below, it starts working out fine.

public override bool CanBuildChannelFactory<TChannel>(BindingContext context)
{
return typeof(TChannel) == typeof(IDuplexChannel);
}
public override IChannelFactory<TChannel> BuildChannelFactory<TChannel>(BindingContext context)
{
return (IChannelFactory<TChannel>)(object)new SizedTcpDuplexChannelFactory(this, context);
}

Now, the duplex part. If everything went right (and by testing the client and server parts separately, we already have a good confidence in the code). I’ll try with a simple duplex contract, as shown below. When the service receives a request (Hello), it will send a series of OnHello messages to the client (more than one).

[ServiceContract(CallbackContract = typeof(ITestCallback))]
public interface ITest
{
[OperationContract]
void Hello(string text);
}
[ServiceContract]
public interface ITestCallback
{
[OperationContract(IsOneWay = true)]
void OnHello(string text);
}
public class Service : ITest
{
public void Hello(string text)
{
Console.WriteLine("[server received] {0}", text);
ITestCallback callback = OperationContext.Current.GetCallbackChannel<ITestCallback>();
ThreadPool.QueueUserWorkItem(delegate
{
for (int i = 1; i <= 5; i++)
callback.OnHello(text + ", " + i);
});
}
}

In order to use a duplex contract in WCF, the client needs to implement the callback interface, then to use the DuplexChannelFactory<TChannel> to create a channel passing that implementation (wrapped in an InstanceContext object).

class Program
{
class ClientCallback : ITestCallback
{
public void OnHello(string text)
{
Console.WriteLine("[client received] {0}", text);
}
}
static void Main(string[] args)
{
string baseAddress = SizedTcpDuplexTransportBindingElement.SizedTcpScheme + "://localhost:8000";
ServiceHost host = new ServiceHost(typeof(Service), new Uri(baseAddress));
Binding binding = new CustomBinding(new SizedTcpDuplexTransportBindingElement());
ServiceEndpoint endpoint = host.AddServiceEndpoint(typeof(ITest), binding, "");
host.Open();
Console.WriteLine("Host opened");
InstanceContext instanceContext = new InstanceContext(new ClientCallback());
EndpointAddress endpointAddress = new EndpointAddress(baseAddress);
DuplexChannelFactory<ITest> factory = new DuplexChannelFactory<ITest>(instanceContext, binding, endpointAddress);
ITest proxy = factory.CreateChannel();
proxy.Hello("John Doe");
Console.WriteLine("Press ENTER to close");
Console.ReadLine();
((IClientChannel)proxy).Close();
factory.Close();
host.Close();
}
}

Run it, and voila, the server is able to send multiple messages to the client without being requested! That’s the magic of a duplex channel in action.

Final thoughts about transport channel extensibility

If there’s one thing which I hope to have shown in those examples, is that WCF is powerful enough to be used even in protocols for which it wasn’t designed. But one thing is certain: creating transport channels should not be your first option (or even your second). There’s a lot of asynchronous code which is inherently error-prone, and there is a lot of “magic” going on over the transport which we really don’t have control over. Often a simple socket-based client (or server) is preferable, but in cases where asynchrony is a plus (e.g., server scenarios, UI threads), you can consider extending WCF with a custom transport, and I hope those posts will shed some light in this area.

Coming up

We’re done with the channel layer, we’ll go back to the service model extensibility in the next post.

[Code in this post]

[Back to the index]

Share via