example/thread-pool/psi.c
static int
handle_transform(INKCont contp)
{
  INKVConn output_conn;
  INKVIO input_vio;
  ContData *data;
  INKIOBufferReader input_reader;
  int toread, avail, psi, toconsume, towrite;
  INKReturnCode retval;

  /* Get the output (downstream) vconnection where we'll write data to. */
  output_conn = INKTransformOutputVConnGet(contp);
  if (output_conn == INK_ERROR_PTR) {
    INKError("[handle_transform] Error while getting transform VC");
    return 1;
  }

  /* Get upstream vio */
  input_vio = INKVConnWriteVIOGet(contp);
  if (input_vio == INK_ERROR_PTR) {
    INKError("[handle_transform] Error while getting input vio");
    return 1;
  }

  data = INKContDataGet(contp);
  INKAssert(data->magic == MAGIC_ALIVE);

  if (!data->output_buffer) {
    data->output_buffer = INKIOBufferCreate();
    data->output_reader = INKIOBufferReaderAlloc(data->output_buffer);

    /* INT_MAX because we don't know yet how much bytes we'll produce */
    data->output_vio = INKVConnWrite(output_conn, contp, data->output_reader, INT_MAX);

    if (data->output_vio == INK_ERROR_PTR) {
      INKError("[handle_transform] Error while writing to downstream VC");
      return 0;
    }
  }

  /* If the input VIO's buffer is NULL, the transformation is over */
  if (!INKVIOBufferGet(input_vio)) {
    INKDebug(DBG_TAG, "input_vio NULL, terminating transformation");
    INKVIONBytesSet(data->output_vio, data->transform_bytes);
    INKVIOReenable(data->output_vio);
    return 1;
  }

  /* Determine how much data we have left to read. */
  toread = INKVIONTodoGet(input_vio);

  if (toread > 0) {
    input_reader = INKVIOReaderGet(input_vio);
    avail = INKIOBufferReaderAvail(input_reader);
    if (avail == INK_ERROR) {
      INKError("[handle_transform] Error while getting number of bytes available");
      return 0;
    }

    /* There are some data available for reading. Let's parse it */
    if (avail > 0) {

      /* No need to parse data if there are too few bytes left to contain
         an include command... */
      if (toread > (PSI_START_TAG_LEN + PSI_END_TAG_LEN)) {
        psi = parse_data(contp, input_reader, avail, &toconsume, &towrite);
      } else {
        towrite = avail;
        toconsume = avail;
        psi = 0;
      }

      if (towrite > 0) {
        /* Update the total size of the doc so far */
        data->transform_bytes += towrite;

        /* Copy the data from the read buffer to the output buffer. */
        retval = INKIOBufferCopy(INKVIOBufferGet(data->output_vio), INKVIOReaderGet(input_vio), towrite, 0);
        if (retval == INK_ERROR) {
          INKError("[handle_transform] Error while copying bytes to output VC");
          return 0;
        }

        /* Reenable the output connection so it can read the data we've produced. */
        retval = INKVIOReenable(data->output_vio);
        if (retval == INK_ERROR) {
          INKError("[handle_transform] Error while reenabling output VC");
          return 0;
        }
      }

      if (toconsume > 0) {
        /* Consume data we've processed an we are no longer interested in */
        retval = INKIOBufferReaderConsume(input_reader, toconsume);
        if (retval == INK_ERROR) {
          INKError("[handle_transform] Error while consuming data from upstream VC");
          return 0;
        }

        /* Modify the input VIO to reflect how much data we've completed. */
        retval = INKVIONDoneSet(input_vio, INKVIONDoneGet(input_vio) + toconsume);
        if (retval == INK_ERROR) {
          INKError("[handle_transform] Error while setting ndone on upstream VC");
          return 0;
        }
      }

      /* Did we find a psi filename to execute in the data ? */
      if (psi) {
        Job *new_job;
        /* Add a request to include a file into the jobs queue.. */
        /* We'll be called back once it's done with an EVENT_IMMEDIATE */
        INKDebug(DBG_TAG, "Psi filename extracted. Adding an include job to thread queue.");
        data->state = STATE_READ_PSI;

        /* Create a new job request and add it to the queue */
        new_job = job_create(contp, &psi_include, NULL);
        add_to_queue(&job_queue, new_job);

        /* Signal to the threads there is a new job */
        thread_signal_job();

        return 1;
      }
    }
  }

  /* Wake up upstream and downstream vconnections */
  wake_up_streams(contp);

  return 1;
}
/*-------------------------------------------------------------------------
  handle_transform
   Get data from upstream vconn.
   Parse it.
   Include file if include tags found.
   Copy data to downstream vconn.
   Wake up upstream to get more data.

  Input:
    contp      continuation for the current transaction
  Output :
  Return Value:
   0 if failure
   1 if success
  -------------------------------------------------------------------------*/
handle_transform
Get data from upstream vconn.
Parse it.
Include file if include tags found.
Copy data to downstream vconn.
Wake up upstream to get more data.

contp      continuation for the current transaction
Output :
Return Value:
0 if failure
1 if success
handle_transform Get data from upstream vconn . Parse it . Include file if include tags found . Copy data to downstream vconn . Wake up upstream to get more data . contp continuation for the current transaction Output : Return Value : 0 if failure 1 if success
static int
handle_transform(INKCont contp)
{
  INKVConn output_conn;
  INKVIO input_vio;
  ContData *data;
  INKIOBufferReader input_reader;
  int toread, avail, psi, toconsume, towrite;
  INKReturnCode retval;
  output_conn = INKTransformOutputVConnGet(contp);
  if (output_conn == INK_ERROR_PTR) {
    INKError("[handle_transform] Error while getting transform VC");
    return 1;
  }
  input_vio = INKVConnWriteVIOGet(contp);
  if (input_vio == INK_ERROR_PTR) {
    INKError("[handle_transform] Error while getting input vio");
    return 1;
  }
  data = INKContDataGet(contp);
  INKAssert(data->magic == MAGIC_ALIVE);
  if (!data->output_buffer) {
    data->output_buffer = INKIOBufferCreate();
    data->output_reader = INKIOBufferReaderAlloc(data->output_buffer);
    data->output_vio = INKVConnWrite(output_conn, contp, data->output_reader, INT_MAX);
    if (data->output_vio == INK_ERROR_PTR) {
      INKError("[handle_transform] Error while writing to downstream VC");
      return 0;
    }
  }
  if (!INKVIOBufferGet(input_vio)) {
    INKDebug(DBG_TAG, "input_vio NULL, terminating transformation");
    INKVIONBytesSet(data->output_vio, data->transform_bytes);
    INKVIOReenable(data->output_vio);
    return 1;
  }
  toread = INKVIONTodoGet(input_vio);
  if (toread > 0) {
    input_reader = INKVIOReaderGet(input_vio);
    avail = INKIOBufferReaderAvail(input_reader);
    if (avail == INK_ERROR) {
      INKError("[handle_transform] Error while getting number of bytes available");
      return 0;
    }
    if (avail > 0) {
      if (toread > (PSI_START_TAG_LEN + PSI_END_TAG_LEN)) {
        psi = parse_data(contp, input_reader, avail, &toconsume, &towrite);
      } else {
        towrite = avail;
        toconsume = avail;
        psi = 0;
      }
      if (towrite > 0) {
        data->transform_bytes += towrite;
        retval = INKIOBufferCopy(INKVIOBufferGet(data->output_vio), INKVIOReaderGet(input_vio), towrite, 0);
        if (retval == INK_ERROR) {
          INKError("[handle_transform] Error while copying bytes to output VC");
          return 0;
        }
        retval = INKVIOReenable(data->output_vio);
        if (retval == INK_ERROR) {
          INKError("[handle_transform] Error while reenabling output VC");
          return 0;
        }
      }
      if (toconsume > 0) {
        retval = INKIOBufferReaderConsume(input_reader, toconsume);
        if (retval == INK_ERROR) {
          INKError("[handle_transform] Error while consuming data from upstream VC");
          return 0;
        }
        retval = INKVIONDoneSet(input_vio, INKVIONDoneGet(input_vio) + toconsume);
        if (retval == INK_ERROR) {
          INKError("[handle_transform] Error while setting ndone on upstream VC");
          return 0;
        }
      }
      if (psi) {
        Job *new_job;
        INKDebug(DBG_TAG, "Psi filename extracted. Adding an include job to thread queue.");
        data->state = STATE_READ_PSI;
        new_job = job_create(contp, &psi_include, NULL);
        add_to_queue(&job_queue, new_job);
        thread_signal_job();
        return 1;
      }
    }
  }
  wake_up_streams(contp);
  return 1;
}
static int handle_transform ( INKCont contp ) { INKVConn output_conn ; INKVIO input_vio ; ContData * data ; INKIOBufferReader input_reader ; int toread , avail , psi , toconsume , towrite ; INKReturnCode retval ; output_conn = INKTransformOutputVConnGet ( contp ) ; if ( output_conn == INK_ERROR_PTR ) { INKError ( " " ) ; return 1 ; } input_vio = INKVConnWriteVIOGet ( contp ) ; if ( input_vio == INK_ERROR_PTR ) { INKError ( " " ) ; return 1 ; } data = INKContDataGet ( contp ) ; INKAssert ( data -> magic == MAGIC_ALIVE ) ; if ( ! data -> output_buffer ) { data -> output_buffer = INKIOBufferCreate ( ) ; data -> output_reader = INKIOBufferReaderAlloc ( data -> output_buffer ) ; data -> output_vio = INKVConnWrite ( output_conn , contp , data -> output_reader , INT_MAX ) ; if ( data -> output_vio == INK_ERROR_PTR ) { INKError ( " " ) ; return 0 ; } } if ( ! INKVIOBufferGet ( input_vio ) ) { INKDebug ( DBG_TAG , " " ) ; INKVIONBytesSet ( data -> output_vio , data -> transform_bytes ) ; INKVIOReenable ( data -> output_vio ) ; return 1 ; } toread = INKVIONTodoGet ( input_vio ) ; if ( toread > 0 ) { input_reader = INKVIOReaderGet ( input_vio ) ; avail = INKIOBufferReaderAvail ( input_reader ) ; if ( avail == INK_ERROR ) { INKError ( " " ) ; return 0 ; } if ( avail > 0 ) { if ( toread > ( PSI_START_TAG_LEN + PSI_END_TAG_LEN ) ) { psi = parse_data ( contp , input_reader , avail , & toconsume , & towrite ) ; } else { towrite = avail ; toconsume = avail ; psi = 0 ; } if ( towrite > 0 ) { data -> transform_bytes += towrite ; retval = INKIOBufferCopy ( INKVIOBufferGet ( data -> output_vio ) , INKVIOReaderGet ( input_vio ) , towrite , 0 ) ; if ( retval == INK_ERROR ) { INKError ( " " ) ; return 0 ; } retval = INKVIOReenable ( data -> output_vio ) ; if ( retval == INK_ERROR ) { INKError ( " " ) ; return 0 ; } } if ( toconsume > 0 ) { retval = INKIOBufferReaderConsume ( input_reader , toconsume ) ; if ( retval == INK_ERROR ) { INKError ( " " ) ; return 0 ; } retval = INKVIONDoneSet ( input_vio , INKVIONDoneGet ( input_vio ) + toconsume ) ; if ( retval == INK_ERROR ) { INKError ( " " ) ; return 0 ; } } if ( psi ) { Job * new_job ; INKDebug ( DBG_TAG , " " ) ; data -> state = STATE_READ_PSI ; new_job = job_create ( contp , & psi_include , NULL ) ; add_to_queue ( & job_queue , new_job ) ; thread_signal_job ( ) ; return 1 ; } } } wake_up_streams ( contp ) ; return 1 ; }
handle_transform
Get data from upstream vconn.
handle_transform Get data from upstream vconn .
/* Get the output (downstream) vconnection where we'll write data to. */ /* Get upstream vio */ /* INT_MAX because we don't know yet how much bytes we'll produce */ /* If the input VIO's buffer is NULL, the transformation is over */ /* Determine how much data we have left to read. */ /* There are some data available for reading. Let's parse it */ /* No need to parse data if there are too few bytes left to contain
         an include command... */ /* Update the total size of the doc so far */ /* Copy the data from the read buffer to the output buffer. */ /* Reenable the output connection so it can read the data we've produced. */ /* Consume data we've processed an we are no longer interested in */ /* Modify the input VIO to reflect how much data we've completed. */ /* Did we find a psi filename to execute in the data ? */ /* Add a request to include a file into the jobs queue.. */ /* We'll be called back once it's done with an EVENT_IMMEDIATE */ /* Create a new job request and add it to the queue */ /* Signal to the threads there is a new job */ /* Wake up upstream and downstream vconnections */