private SendReturn send(int messageType,String message) throws InterruptedException, ExtractorException {
  String length=String.format("%d",message.length());
  String lenlen=String.format("%d",length.length());
  String type=String.format("%04d",messageType);
  String outMessage;
  int bytesRead;
  String replyString;
  int ll;
  int t;
  SendReturn retval=new SendReturn();
  boolean lengthKnown;
  int totalBytesRead=0;
  int totalBytesNeeded;
  boolean longMessageSupport=true;
  if (!connected)   throw new OracleExtractException("Not connected");
  outMessage=lenlen + length + type+ message;
  if (logger.isDebugEnabled()) {
    logger.debug("Sending Message type " + messageType);
    logger.debug(" - Message is -> " + message);
    logger.debug(" - Therefore sending -> " + outMessage);
  }
  out.println(outMessage);
  if (messageType != Const.MessageControl) {
    try {
      if (longMessageSupport) {
        lengthKnown=false;
        totalBytesRead=0;
        totalBytesNeeded=0;
        for (; ; ) {
          bytesRead=in.read(replyBuffer,totalBytesRead,replyBufferSize - totalBytesRead - 1);
          if (bytesRead == -1) {
            throw new OracleExtractException("End of Stream");
          }
          totalBytesRead+=bytesRead;
          logger.debug("Read = " + bytesRead + " total = "+ totalBytesRead+ " needed = "+ totalBytesNeeded);
          if (lengthKnown) {
            if (totalBytesRead >= totalBytesNeeded)             break;
 else             continue;
          }
          if (totalBytesRead <= 9)           continue;
          lengthKnown=true;
          replyBuffer[totalBytesRead]='\0';
          lengthBuffer[0]=replyBuffer[0];
          lengthBuffer[1]=replyBuffer[1];
          lengthBuffer[2]=replyBuffer[2];
          lengthBuffer[3]=replyBuffer[3];
          lengthBuffer[4]=replyBuffer[4];
          lengthBuffer[5]=replyBuffer[5];
          lengthBuffer[6]=replyBuffer[6];
          lengthBuffer[7]=replyBuffer[7];
          lengthBuffer[8]=replyBuffer[8];
          lengthBuffer[9]=replyBuffer[9];
          lengthBuffer[10]='\0';
          String lengthString=new String(lengthBuffer);
          totalBytesNeeded=Integer.parseInt(lengthString.substring(1,9),16);
          totalBytesNeeded+=13;
          if (totalBytesNeeded + 1 > replyBufferSize)           increaseReplyBuffer(totalBytesNeeded + 1);
          if (totalBytesRead >= totalBytesNeeded)           break;
        }
      }
 else {
        totalBytesRead=in.read(replyBuffer);
      }
    }
 catch (    ClosedByInterruptException e) {
      throw new InterruptedException("Oracle extractor was interrupted");
    }
catch (    IOException e) {
      throw new OracleExtractException("End of Stream");
    }
    if (totalBytesRead == -1) {
      throw new OracleExtractException("End of Stream");
    }
    replyBuffer[totalBytesRead]='\0';
    replyString=new String(replyBuffer,0,totalBytesRead);
    if (logger.isDebugEnabled()) {
      logger.debug("Reply = \"" + replyString + "\"\n");
    }
    ll=Integer.parseInt(replyString.substring(0,1));
    t=Integer.parseInt(replyString.substring(1 + ll,1 + ll + 4));
    replyString=replyString.substring(1 + ll + 4);
    if (logger.isDebugEnabled()) {
      logger.debug("Reply = \"" + replyString + "\"\n");
    }
    retval.type=t;
    retval.reply=replyString;
  }
 else {
    retval.type=Const.ReplyReserved;
    retval.reply=new String("Empty");
  }
  if (retval.type == Const.ReplyError)   throw new OracleExtractException("Error from C Oracle Extractor");
  msgCount++;
  return retval;
}

private SendReturn send ( int messageType , String message ) throws InterruptedException , ExtractorException { String length = String . format ( "%d" , message . length ( ) ) ; String lenlen = String . format ( "%d" , length . length ( ) ) ; String type = String . format ( "%04d" , messageType ) ; String outMessage ; int bytesRead ; String replyString ; int ll ; int t ; SendReturn retval = new SendReturn ( ) ; boolean lengthKnown ; int totalBytesRead = 0 ; int totalBytesNeeded ; boolean longMessageSupport = true ; if ( ! connected ) throw new OracleExtractException ( "Not connected" ) ; outMessage = lenlen + length + type + message ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Sending Message type " + messageType ) ; logger . debug ( " - Message is -> " + message ) ; logger . debug ( " - Therefore sending -> " + outMessage ) ; } out . println ( outMessage ) ; if ( messageType != Const . MessageControl ) { try { if ( longMessageSupport ) { lengthKnown = false ; totalBytesRead = 0 ; totalBytesNeeded = 0 ; for ( ; ; ) { bytesRead = in . read ( replyBuffer , totalBytesRead , replyBufferSize - totalBytesRead - 1 ) ; if ( bytesRead == - 1 ) { throw new OracleExtractException ( "End of Stream" ) ; } totalBytesRead += bytesRead ; logger . debug ( "Read = " + bytesRead + " total = " + totalBytesRead + " needed = " + totalBytesNeeded ) ; if ( lengthKnown ) { if ( totalBytesRead >= totalBytesNeeded ) break ; else continue ; } if ( totalBytesRead <= 9 ) continue ; lengthKnown = true ; replyBuffer [ totalBytesRead ] = '\0' ; lengthBuffer [ 0 ] = replyBuffer [ 0 ] ; lengthBuffer [ 1 ] = replyBuffer [ 1 ] ; lengthBuffer [ 2 ] = replyBuffer [ 2 ] ; lengthBuffer [ 3 ] = replyBuffer [ 3 ] ; lengthBuffer [ 4 ] = replyBuffer [ 4 ] ; lengthBuffer [ 5 ] = replyBuffer [ 5 ] ; lengthBuffer [ 6 ] = replyBuffer [ 6 ] ; lengthBuffer [ 7 ] = replyBuffer [ 7 ] ; lengthBuffer [ 8 ] = replyBuffer [ 8 ] ; lengthBuffer [ 9 ] = replyBuffer [ 9 ] ; lengthBuffer [ 10 ] = '\0' ; String lengthString = new String ( lengthBuffer ) ; totalBytesNeeded = Integer . parseInt ( lengthString . substring ( 1 , 9 ) , 16 ) ; totalBytesNeeded += 13 ; if ( totalBytesNeeded + 1 > replyBufferSize ) increaseReplyBuffer ( totalBytesNeeded + 1 ) ; if ( totalBytesRead >= totalBytesNeeded ) break ; } } else { totalBytesRead = in . read ( replyBuffer ) ; } } catch ( ClosedByInterruptException e ) { throw new InterruptedException ( "Oracle extractor was interrupted" ) ; } catch ( IOException e ) { throw new OracleExtractException ( "End of Stream" ) ; } if ( totalBytesRead == - 1 ) { throw new OracleExtractException ( "End of Stream" ) ; } replyBuffer [ totalBytesRead ] = '\0' ; replyString = new String ( replyBuffer , 0 , totalBytesRead ) ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Reply = \"" + replyString + "\"\n" ) ; } ll = Integer . parseInt ( replyString . substring ( 0 , 1 ) ) ; t = Integer . parseInt ( replyString . substring ( 1 + ll , 1 + ll + 4 ) ) ; replyString = replyString . substring ( 1 + ll + 4 ) ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Reply = \"" + replyString + "\"\n" ) ; } retval . type = t ; retval . reply = replyString ; } else { retval . type = Const . ReplyReserved ; retval . reply = new String ( "Empty" ) ; } if ( retval . type == Const . ReplyError ) throw new OracleExtractException ( "Error from C Oracle Extractor" ) ; msgCount ++ ; return retval ; }
send message to dslisten - typically an extract next change message
