Hi,
While playing around with the server-transform plugin example, i noticed it
was broken. After some meddling around, i got it working again.
The changes are mainly that in some cases content-length was passed in
network-byte order where it shouldn't be.
Another fix was handling the TS_EVENT_IMMEDIATE event
in transform_write_event(). i currently perform
a TSVIOReenable(data->server_vio); on that event, and it seems to work.
However, I'm not sure at all if that is the correct thing to do, so if
anybody could help me out on that? :-).
I have tested the changed code with a custom echo server, and confirmed that
the plugin is actually sending and receiving to the echo server.
I added the working code below.
Regards,
Otto van der Schaaf
#include <string.h>
#include <stdio.h>
#if !defined (_WIN32)
# include <netinet/in.h>
#else
# include <windows.h>
#endif
#include <ts/ts.h>
#define STATE_BUFFER 1
#define STATE_CONNECT 2
#define STATE_WRITE 3
#define STATE_READ_STATUS 4
#define STATE_READ 5
#define STATE_BYPASS 6
typedef struct
{
int state;
TSHttpTxn txn;
TSIOBuffer input_buf;
TSIOBufferReader input_reader;
TSIOBuffer output_buf;
TSIOBufferReader output_reader;
TSVConn output_vc;
TSVIO output_vio;
TSAction pending_action;
TSVConn server_vc;
TSVIO server_vio;
int content_length;
} TransformData;
static TSCont transform_create(TSHttpTxn txnp);
static void transform_destroy(TSCont contp);
static int transform_connect(TSCont contp, TransformData * data);
static int transform_write(TSCont contp, TransformData * data);
static int transform_read_status(TSCont contp, TransformData * data);
static int transform_read(TSCont contp, TransformData * data);
static int transform_bypass(TSCont contp, TransformData * data);
static int transform_buffer_event(TSCont contp, TransformData * data,
TSEvent event, void *edata);
static int transform_connect_event(TSCont contp, TransformData * data,
TSEvent event, void *edata);
static int transform_write_event(TSCont contp, TransformData * data, TSEvent
event, void *edata);
static int transform_read_status_event(TSCont contp, TransformData * data,
TSEvent event, void *edata);
static int transform_read_event(TSCont contp, TransformData * data, TSEvent
event, void *edata);
static int transform_bypass_event(TSCont contp, TransformData * data,
TSEvent event, void *edata);
static int transform_handler(TSCont contp, TSEvent event, void *edata);
#if !defined (_WIN32)
static in_addr_t server_ip;
#else
static unsigned int server_ip;
#endif
static int server_port;
static TSCont transform_create(TSHttpTxn txnp)
{
TSCont contp;
TransformData *data;
contp = TSTransformCreate(transform_handler, txnp);
data = (TransformData *) TSmalloc(sizeof(TransformData));
data->state = STATE_BUFFER;
data->txn = txnp;
data->input_buf = NULL;
data->input_reader = NULL;
data->output_buf = NULL;
data->output_reader = NULL;
data->output_vio = NULL;
data->output_vc = NULL;
data->pending_action = NULL;
data->server_vc = NULL;
data->server_vio = NULL;
data->content_length = 0;
TSContDataSet(contp, data);
return contp;
}
static void transform_destroy(TSCont contp)
{
TransformData *data;
data = TSContDataGet(contp);
if (data != NULL) {
if (data->input_buf)
TSIOBufferDestroy(data->input_buf);
if (data->output_buf)
TSIOBufferDestroy(data->output_buf);
if (data->pending_action)
TSActionCancel(data->pending_action);
if (data->server_vc)
TSVConnAbort(data->server_vc, 1);
TSfree(data);
} else {
TSError("Unable to get Continuation's Data. TSContDataGet returns
NULL");
}
TSContDestroy(contp);
}
static int transform_connect(TSCont contp, TransformData * data)
{
TSDebug ("strans", "@@@ transform_connect\n");
TSAction action;
int content_length;
struct sockaddr_in ip_addr;
data->state = STATE_CONNECT;
content_length = TSIOBufferReaderAvail(data->input_reader);
TSDebug ("strans", "@@@ transform_connect: content length
%i\n",content_length);
if (content_length >= 0) {
data->content_length = content_length;
data->content_length = htonl(data->content_length);
TSDebug ("strans", "@@@ transform_connect: htonl / data->content_length
%i\n",data->content_length);
/* Prepend the content length to the buffer.
* If we decide to not send the content to the transforming
* server then we need to make sure and skip input_reader
* over the content length.
*/
{
TSIOBuffer temp;
TSIOBufferReader tempReader;
temp = TSIOBufferCreate();
tempReader = TSIOBufferReaderAlloc(temp);
TSDebug ("strans", "@@@ tmp reader created and allocated\n");
int TSIOBufferWriteRes = TSIOBufferWrite(temp, (const char *)
&content_length, sizeof(int));
TSDebug ("strans", "@@@ TSIOBufferWrite / wrote %i bytes for content
length\n",TSIOBufferWriteRes);
TSDebug ("strans", "@@@ TSIOBufferWrite / wrote content length\n");
int TSIOBufferCopyres = TSIOBufferCopy(temp, data->input_reader,
content_length, 0);
TSDebug ("strans", "@@@ TSIOBufferWrite / wrote %i
bytes\n",TSIOBufferCopyres);
TSDebug ("strans", "@@@ TSIOBufferCopy executed\n");
TSIOBufferReaderFree(data->input_reader);
TSIOBufferDestroy(data->input_buf);
data->input_buf = temp;
data->input_reader = tempReader;
}
} else {
TSError("TSIOBufferReaderAvail returns TS_ERROR");
return 0;
}
/* TODO: This only supports IPv4, probably should be changed at some
point, but
it's an example ... */
memset(&ip_addr, 0, sizeof(ip_addr));
ip_addr.sin_family = AF_INET;
ip_addr.sin_addr.s_addr = server_ip; /* Should be in network byte order */
ip_addr.sin_port = server_port;
action = TSNetConnect(contp, (struct sockaddr const*)&ip_addr);
if (!TSActionDone(action)) {
data->pending_action = action;
}
return 0;
}
static int transform_write(TSCont contp, TransformData * data)
{
TSDebug ("strans", "@@@ transform_write\n");
int content_length;
data->state = STATE_WRITE;
content_length = TSIOBufferReaderAvail(data->input_reader);
if (content_length >= 0) {
TSDebug ("strans", "@@@ transform_write - %i bytes\n", content_length);
data->server_vio = TSVConnWrite(data->server_vc, contp,
TSIOBufferReaderClone(data->input_reader), content_length);
TSDebug ("strans", "@@@ transform_write - data_server_vio: %i \n",
data->server_vio);
} else {
TSError("TSIOBufferReaderAvail returns TS_ERROR");
}
return 0;
}
static int transform_read_status(TSCont contp, TransformData * data)
{
TSDebug ("strans", "@@@ transform_read_status\n");
data->state = STATE_READ_STATUS;
data->output_buf = TSIOBufferCreate();
data->output_reader = TSIOBufferReaderAlloc(data->output_buf);
if (data->output_reader != NULL) {
data->server_vio = TSVConnRead(data->server_vc, contp, data->output_buf,
sizeof(int));
} else {
TSError("Error in Allocating a Reader to output buffer.
TSIOBufferReaderAlloc returns NULL");
}
return 0;
}
static int transform_read(TSCont contp, TransformData * data)
{
TSDebug ("strans", "@@@ transform_read\n");
data->state = STATE_READ;
TSIOBufferDestroy(data->input_buf);
data->input_buf = NULL;
data->input_reader = NULL;
data->server_vio = TSVConnRead(data->server_vc, contp, data->output_buf,
data->content_length);
data->output_vc = TSTransformOutputVConnGet((TSVConn) contp);
if (data->output_vc == NULL) {
TSError("TSTransformOutputVConnGet returns NULL");
} else {
data->output_vio = TSVConnWrite(data->output_vc, contp,
data->output_reader, data->content_length);
if (data->output_vio == NULL) {
TSError("TSVConnWrite returns NULL");
}
}
return 0;
}
static int transform_bypass(TSCont contp, TransformData * data)
{
TSDebug ("strans", "@@@ transform_bypass\n");
data->state = STATE_BYPASS;
if (data->server_vc) {
TSVConnAbort(data->server_vc, 1);
data->server_vc = NULL;
data->server_vio = NULL;
}
if (data->output_buf) {
TSIOBufferDestroy(data->output_buf);
data->output_buf = NULL;
data->output_reader = NULL;
}
TSIOBufferReaderConsume(data->input_reader, sizeof(int));
data->output_vc = TSTransformOutputVConnGet((TSVConn) contp);
if (data->output_vc == NULL) {
TSError("TSTransformOutputVConnGet returns NULL");
} else {
data->output_vio =
TSVConnWrite(data->output_vc, contp, data->input_reader,
TSIOBufferReaderAvail(data->input_reader));
if (data->output_vio == NULL) {
TSError("TSVConnWrite returns NULL");
}
}
return 1;
}
static int transform_buffer_event(TSCont contp, TransformData * data,
TSEvent event, void *edata)
{
TSDebug ("strans", "@@@ transform_buffer_event\n");
TSVIO write_vio;
int towrite;
int avail;
if (!data->input_buf) {
data->input_buf = TSIOBufferCreate();
data->input_reader = TSIOBufferReaderAlloc(data->input_buf);
}
/* Get the write VIO for the write operation that was performed on
ourself. This VIO contains the buffer that we are to read from
as well as the continuation we are to call when the buffer is
empty. */
write_vio = TSVConnWriteVIOGet(contp);
/* We also check to see if the write VIO's buffer is non-NULL. A
NULL buffer indicates that the write operation has been
shutdown and that the continuation does not want us to send any
more WRITE_READY or WRITE_COMPLETE events. For this buffered
transformation that means we're done buffering data. */
if (!TSVIOBufferGet(write_vio)) {
return transform_connect(contp, data);
}
/* Determine how much data we have left to read. For this server
transform plugin this is also the amount of data we have left
to write to the output connection. */
towrite = TSVIONTodoGet(write_vio);
if (towrite > 0) {
/* The amount of data left to read needs to be truncated by
the amount of data actually in the read buffer. */
avail = TSIOBufferReaderAvail(TSVIOReaderGet(write_vio));
if (towrite > avail) {
towrite = avail;
}
if (towrite > 0) {
/* Copy the data from the read buffer to the input buffer. */
TSIOBufferCopy(data->input_buf, TSVIOReaderGet(write_vio), towrite,
0);
/* Tell the read buffer that we have read the data and are no
longer interested in it. */
TSIOBufferReaderConsume(TSVIOReaderGet(write_vio), towrite);
/* Modify the write VIO to reflect how much data we've
completed. */
TSVIONDoneSet(write_vio, TSVIONDoneGet(write_vio) + towrite);
}
}
/* Now we check the write VIO to see if there is data left to
read. */
if (TSVIONTodoGet(write_vio) > 0) {
/* Call back the write VIO continuation to let it know that we
are ready for more data. */
TSContCall(TSVIOContGet(write_vio), TS_EVENT_VCONN_WRITE_READY,
write_vio);
} else {
/* Call back the write VIO continuation to let it know that we
have completed the write operation. */
TSContCall(TSVIOContGet(write_vio), TS_EVENT_VCONN_WRITE_COMPLETE,
write_vio);
/* start compression... */
return transform_connect(contp, data);
}
return 0;
}
static int transform_connect_event(TSCont contp, TransformData * data,
TSEvent event, void *edata)
{
TSDebug ("strans", "@@@ transform_connect_event\n");
switch (event) {
case TS_EVENT_NET_CONNECT:
TSDebug ("strans", "@@@ transform_connect_event -
TS_EVENT_NET_CONNECT\n");
data->pending_action = NULL;
data->server_vc = (TSVConn) edata;
return transform_write(contp, data);
case TS_EVENT_NET_CONNECT_FAILED:
TSDebug ("strans", "@@@ transform_connect_event -
TS_EVENT_NET_CONNECT_FAILED\n");
data->pending_action = NULL;
return transform_bypass(contp, data);
default:
TSDebug ("strans", "@@@ transform_connect_event - default\n");
break;
}
return 0;
}
static int transform_write_event(TSCont contp, TransformData * data, TSEvent
event, void *edata)
{
TSDebug ("strans", "@@@ transform_write_event - WAAH2\n");
switch (event) {
case TS_EVENT_VCONN_WRITE_READY:
TSDebug ("strans", "@@@ transform_write_event -
S_EVENT_VCONN_WRITE_READY\n");
TSVIOReenable(data->server_vio);
break;
case TS_EVENT_VCONN_WRITE_COMPLETE:
TSDebug ("strans", "@@@ transform_write_event -
TS_EVENT_VCONN_WRITE_COMPLETE\n");
return transform_read_status(contp, data);
case TS_EVENT_ERROR:
TSDebug ("strans", "@@@ transform_write_event - Explicit
transform_write_event error. Close down and bypass\n");
return transform_bypass(contp, data);
case TS_EVENT_IMMEDIATE:
TSDebug ("strans", "@@@ transform_write_event - IMMEDIATE\n");
TSVIOReenable(data->server_vio);
break;
default:
/* An error occurred while writing to the server. Close down the
connection to the server and bypass. */
TSDebug ("strans", "@@@ transform_write_event - An error occurred while
writing to the server:%d. Close down/bypass\n",event);
return transform_bypass(contp, data);
}
return 0;
}
static int transform_read_status_event(TSCont contp, TransformData * data,
TSEvent event, void *edata)
{
TSDebug ("strans", "@@@ transform_read_status_event\n");
switch (event) {
case TS_EVENT_ERROR:
case TS_EVENT_VCONN_EOS:
TSDebug ("strans", "@@@ transform_read_status_event / TS_EVENT_ERROR |
TS_EVENT_VCONN_EOS\n");
return transform_bypass(contp, data);
case TS_EVENT_VCONN_READ_COMPLETE:
TSDebug ("strans", "@@@ transform_read_status_event /
TS_EVENT_VCONN_READ_COMPLETE\n");
if (TSIOBufferReaderAvail(data->output_reader) == sizeof(int)) {
TSIOBufferBlock blk;
char *buf;
void *buf_ptr;
int64_t avail;
int64_t read_nbytes = sizeof(int);
int64_t read_ndone = 0;
TSDebug ("strans", "@@@ transform_read_status_event %d
bytes\n",&data->content_length);
buf_ptr = &data->content_length;
while (read_nbytes > 0) {
TSDebug ("strans", "@@@ transform_read_status_event read_nbytes:
%d\n",read_nbytes);
blk = TSIOBufferReaderStart(data->output_reader);
buf = (char *) TSIOBufferBlockReadStart(blk, data->output_reader,
&avail);
read_ndone = (avail >= read_nbytes) ? read_nbytes : avail;
memcpy(buf_ptr, buf, read_ndone);
if (read_ndone > 0) {
TSIOBufferReaderConsume(data->output_reader, read_ndone);
read_nbytes -= read_ndone;
/* move ptr frwd by read_ndone bytes */
buf_ptr = (char *) buf_ptr + read_ndone;
}
}
//data->content_length = ntohl(data->content_length);
//data->content_length = ntohl(data->content_length);
TSDebug ("strans", "@@@ transform_read_status_event
data->content_length: %d\n", data->content_length);
return transform_read(contp, data);
}
TSDebug ("strans", "@@@ transform_read_status_event /
TS_EVENT_VCONN_READ_COMPLETE -> return bypass?\n");
return transform_bypass(contp, data);
default:
break;
}
return 0;
}
static int transform_read_event(TSCont contp, TransformData * data, TSEvent
event, void *edata)
{
TSDebug ("strans", "@@@ transform_read_event\n");
switch (event) {
case TS_EVENT_ERROR:
TSDebug ("strans", "@@@ transform_read_event / TS_EVENT_ERROR\n");
TSVConnAbort(data->server_vc, 1);
data->server_vc = NULL;
data->server_vio = NULL;
TSVConnAbort(data->output_vc, 1);
data->output_vc = NULL;
data->output_vio = NULL;
break;
case TS_EVENT_VCONN_EOS:
TSDebug ("strans", "@@@ transform_read_event / TS_EVENT_VCONN_EOS\n");
TSVConnAbort(data->server_vc, 1);
data->server_vc = NULL;
data->server_vio = NULL;
TSVConnAbort(data->output_vc, 1);
data->output_vc = NULL;
data->output_vio = NULL;
break;
case TS_EVENT_VCONN_READ_COMPLETE:
TSDebug ("strans", "@@@ transform_read_event /
TS_EVENT_VCONN_READ_COMPLETE\n");
TSVConnClose(data->server_vc);
data->server_vc = NULL;
data->server_vio = NULL;
TSVIOReenable(data->output_vio);
break;
case TS_EVENT_VCONN_READ_READY:
TSDebug ("strans", "@@@ transform_read_event /
TS_EVENT_VCONN_READ_READY\n");
TSVIOReenable(data->output_vio);
break;
case TS_EVENT_VCONN_WRITE_COMPLETE:
TSDebug ("strans", "@@@ transform_read_event /
TS_EVENT_VCONN_WRITE_COMPLETE\n");
TSVConnShutdown(data->output_vc, 0, 1);
break;
case TS_EVENT_VCONN_WRITE_READY:
TSDebug ("strans", "@@@ transform_read_event /
TS_EVENT_VCONN_WRITE_READY\n");
TSVIOReenable(data->server_vio);
break;
default:
break;
}
return 0;
}
static int transform_bypass_event(TSCont contp, TransformData * data,
TSEvent event, void *edata)
{
TSDebug ("strans", "@@@ transform_bypass_event\n");
switch (event) {
case TS_EVENT_VCONN_WRITE_COMPLETE:
TSDebug ("strans", "@@@ transform_bypass_event /
TS_EVENT_VCONN_WRITE_COMPLETE\n");
TSVConnShutdown(data->output_vc, 0, 1);
break;
case TS_EVENT_VCONN_WRITE_READY:
TSDebug ("strans", "@@@ transform_bypass_event /
TS_EVENT_VCONN_WRITE_READY\n");
default:
TSDebug ("strans", "@@@ transform_bypass_event / default\n");
TSVIOReenable(data->output_vio);
break;
}
return 0;
}
static int transform_handler(TSCont contp, TSEvent event, void *edata)
{
TSDebug ("strans", "@@@ transform_handler\n");
/* Check to see if the transformation has been closed by a call to
TSVConnClose. */
if (TSVConnClosedGet(contp)) {
TSDebug ("strans", "@@@ transform_handler -> TSVConnClosedGet -> destroy
contp\n");
transform_destroy(contp);
return 0;
} else {
TransformData *data;
int val = 0;
data = (TransformData *)TSContDataGet(contp);
if (data == NULL) {
TSError("Didn't get Continuation's Data. Ignoring Event..");
return 0;
}
do {
switch (data->state) {
case STATE_BUFFER:
val = transform_buffer_event(contp, data, event, edata);
break;
case STATE_CONNECT:
val = transform_connect_event(contp, data, event, edata);
break;
case STATE_WRITE:
val = transform_write_event(contp, data, event, edata);
break;
case STATE_READ_STATUS:
val = transform_read_status_event(contp, data, event, edata);
break;
case STATE_READ:
val = transform_read_event(contp, data, event, edata);
break;
case STATE_BYPASS:
val = transform_bypass_event(contp, data, event, edata);
break;
}
} while (val);
}
return 0;
}
static int request_ok(TSHttpTxn txnp)
{
/* Is the initial client request OK for transformation. This is a
good place to check accept headers to see if the client can
accept a transformed document. */
return 1;
}
static int cache_response_ok(TSHttpTxn txnp)
{
/* Is the response we're reading from cache OK for
* transformation. This is a good place to check the cached
* response to see if it is transformable. The default
* behavior is to cache transformed content; therefore
* to avoid transforming twice we will not transform
* content served from the cache.
*/
return 0;
}
static int server_response_ok(TSHttpTxn txnp)
{
/* Is the response the server sent OK for transformation. This is
* a good place to check the server's response to see if it is
* transformable. In this example, we will transform only "200 OK"
* responses.
*/
TSDebug ("strans", "@@@ server_response_ok\n");
TSMBuffer bufp;
TSMLoc hdr_loc;
TSHttpStatus resp_status;
if (TSHttpTxnServerRespGet(txnp, &bufp, &hdr_loc) != TS_SUCCESS) {
TSError("Unable to get handle to Server Response");
return 0;
}
resp_status = TSHttpHdrStatusGet(bufp, hdr_loc);
if (TS_HTTP_STATUS_OK == resp_status) {
if (TSHandleMLocRelease(bufp, TS_NULL_MLOC, hdr_loc) != TS_SUCCESS) {
TSError("Unable to release handle to server request");
}
return 1;
} else {
if (TSHandleMLocRelease(bufp, TS_NULL_MLOC, hdr_loc) != TS_SUCCESS) {
TSError("Unable to release handle to server request");
}
return 0;
}
}
static int transform_plugin(TSCont contp, TSEvent event, void *edata)
{
TSDebug ("strans", "@@@ transform_plugin\n");
TSHttpTxn txnp = (TSHttpTxn) edata;
switch (event) {
case TS_EVENT_HTTP_READ_REQUEST_HDR:
if (request_ok(txnp)) {
TSHttpTxnHookAdd(txnp, TS_HTTP_READ_CACHE_HDR_HOOK, contp);
TSHttpTxnHookAdd(txnp, TS_HTTP_READ_RESPONSE_HDR_HOOK, contp);
}
TSHttpTxnReenable(txnp, TS_EVENT_HTTP_CONTINUE);
break;
case TS_EVENT_HTTP_READ_CACHE_HDR:
if (cache_response_ok(txnp)) {
TSHttpTxnHookAdd(txnp, TS_HTTP_RESPONSE_TRANSFORM_HOOK,
transform_create(txnp));
}
TSHttpTxnReenable(txnp, TS_EVENT_HTTP_CONTINUE);
break;
case TS_EVENT_HTTP_READ_RESPONSE_HDR:
if (server_response_ok(txnp)) {
TSHttpTxnHookAdd(txnp, TS_HTTP_RESPONSE_TRANSFORM_HOOK,
transform_create(txnp));
}
TSHttpTxnReenable(txnp, TS_EVENT_HTTP_CONTINUE);
break;
default:
break;
}
return 0;
}
int check_ts_version()
{
const char *ts_version = TSTrafficServerVersionGet();
int result = 0;
if (ts_version) {
int major_ts_version = 0;
int minor_ts_version = 0;
int patch_ts_version = 0;
if (sscanf(ts_version, "%d.%d.%d", &major_ts_version, &minor_ts_version,
&patch_ts_version) != 3) {
return 0;
}
/* Since this is an TS-SDK 2.0 plugin, we need at least Traffic Server
2.0 to run */
if (major_ts_version >= 2) {
result = 1;
}
}
return result;
}
void
TSPluginInit(int argc, const char *argv[])
{
TSPluginRegistrationInfo info;
TSCont cont;
info.plugin_name = "transform test";
info.vendor_name = "nobody";
info.support_email = "osch...@gmail.com";
TSDebug ("strans", "hello from plugin\n");
if (TSPluginRegister(TS_SDK_VERSION_3_0, &info) != TS_SUCCESS) {
TSError("Plugin registration failed.\n");
}
if (!check_ts_version()) {
TSError("Plugin requires Traffic Server 3.0 or later\n");
return;
}
/* connect to the echo port on localhost */
server_ip = (127 << 24) | (0 << 16) | (0 << 8) | (1);
server_ip = htonl(server_ip);
server_port = 7;
cont = TSContCreate(transform_plugin, NULL);
TSHttpHookAdd(TS_HTTP_READ_REQUEST_HDR_HOOK, cont);
}
