RequestDelegate problems

20 views

Skip to first unread message

David Martin

unread,

Jun 15, 2011, 9:10:14 AM6/15/11

to Kayak HTTP

Hiya, I am tring to get to the requestBody, in the examples.
(RequestDelegate : IHttpRequestDelegate) from Kayak "0.6.1.0" taken
from git
It is obvious i am missing something, Could somebody enlighten me?
Thanks

Benjamin van der Veen

unread,

Jun 15, 2011, 1:08:19 PM6/15/11

to kayak...@googlegroups.com

The request and response bodies are represented by IDataProducer, and
the data within them is provided to an IDataConsumer. See the
definition of these interfaces near the bottom here:
https://github.com/kayak/kayak/blob/3d2cf9dd92bde313bd1f3c52a84b6f5833547a46/Kayak/Net/Net.cs

These interfaces are analogous to IObservable and IObserver, and in
fact, can be used in exactly the same way if you just ignore the
return value and 'continuation' parameter of the OnData callback.

So, for example, if you wanted to buffer the request body and get it
as a string, you could do something like this (I'm just typing this
out freehand, so watch out for compile errors, but it should
illustrate the concept):

class BufferingConsumer : IDataConsumer
{
List<ArraySegment<byte>> buffer = new List<ArraySegment<byte>>();
Action<string> resultCallback;
Action<Exception> errorCallback;

public BufferingConsumer(Action<string> resultCallback,
Action<Exception> errorCallback) {
this.resultCallback = resultCallback;
this.errorCallback = errorCallback;
}

bool OnData(ArraySegment<byte> data, Action continuation) {
// since we're just buffering, ignore the continuation.
// TODO: place an upper limit on the size of the buffer.
// don't want a client to take up all the RAM on our server!
buffer.Add(data);
return false;
}

void OnError(Exception error) {
errorCallback(error);
}

void OnEnd() {
// turn the buffer into a string.
//
// (if this isn't what you want, you could skip
// this step and make the result callback accept
// List<ArraySegment<byte>> or whatever)
//
var str = buffer.Select(b =>
Encoding.UTF8.GetString(b)).Aggregate((result, next) => result +
next);

resultCallback(str);
}
}

Then you could make a sweet extension method:

public static IDisposable AsString(IDataProducer producer,
Action<string> callback, Action<Exception> error) {
return producer.Subscribe(new BufferingConsumer(callback, error));
}

The IDisposable is used to cancel the asynchronous operation, just
like with IObservable.

In C# 5, it would be trivial to adapt this AsString method to work
with the async/await syntax.