Lesen und Senden von Daten

/*!\brief Read answer from RRI server
*
* Use this function to read an answer from the RRI-Server
*
* @param ssl SSL context
* @return Pointer to a c-string, which contains the answer. The memory used for this
* string is allocated with malloc and must be freed by the application with free().
* If an error occured, the function returns NULL.
*/
char *rriRead(SSL *ssl)
{
    char *buf;
    int s,nl,size,rest;
    nl=0;
    size=0;
    rest=4;
    while (rest) {
        s=SSL_read(ssl,(char*)&nl+(4-rest),rest);
        if (s<=0) return (NULL);                  // Fehler aufgetreten
        rest-=s;
    }
    size=ntohl(nl);
    buf=(char*)malloc(size+2);
    if (!buf) {
        // Not enough memory
        return (NULL);
    }
    rest=size;
    while (rest) {
        s=SSL_read(ssl,(char*)buf+(size-rest),rest);
        if (s<=0) {
           free(buf);
           return (0);                            // Fehler aufgetreten
        }
       rest-=s;
    }
    buf[size]=0;
    return (buf);
}

/*!\brief Send order to RRI server
*
* Use this function to send an order to the RRI server
*
* @param ssl SSL context
* @param order C-String, which contains the UTF-8 encoded order
* @return Returns the length of the order, if it was successful send to the
* RRI server or 0, if an error occured.
*/
int rriSend(SSL *ssl, const char *order)
{
  int size,nl,s;
  size=(int)strlen(order);
  nl=htonl(size);
  // Send 4-Byte perfix with length of order in bytes
  s  =SSL_write(ssl,(char*)&nl,4);
  if (s!=4) {
      // Error
      return (0);
  }
  // Send order
  s=SSL_write(ssl,order,size);
  if (s!=size) {
      // Error
      return (0);
  }
  return (s);
}

/*!\brief Read answer from RRI-Server
*
* \param $socket Socket handle
* \return Returns string with answer from RRI server or false, if an error
* occured.
*/
function rriReadData($socket)
{
    // Step 1: read 4-Byte RRI-Header
    $nlen="";
    $rest=4;
    while ($rest) {
        $a=fread($socket,$rest);               // read answer
        $bytesread=strlen($a);
        $nlen.=$a;
        $rest-=$bytesread;
        if (feof($socket)) return false;
    }

    $len=rriUnpack($nlen);                // convert bytes to local order

    // Step 2: read payload
    $rest=$len;
    $answer="";
    while ($rest) {
        $a=fread($socket,$rest);               // read answer
        $bytesread=strlen($a);
        $answer.=$a;
        $rest-=$bytesread;
        if (feof($socket)) return false;
    }
    return $answer;
}

/*!\brief Converts network byte order to local byte order
*
* \param $value Integer in network byte order
* \return Returns the converted $value in local byte order
*/
function rriUnpack($value)
{
    $a[0]=ord($value[0]);
    $a[1]=ord($value[1]);
    $a[2]=ord($value[2]);
    $a[3]=ord($value[3]);
    return $a[3]+($a[2]<<8) + ($a[1]<<16) + ($a[0]<<24);
}

/*!\brief Sends order to RRI-Server
*
* \param $socket Socket handle
* \param $order String, which should be send to the RRI-server
* \return Returns 1 on success, 0 on failure
*/
function rriSendData($socket, $order)
{
    $len=strlen($order);
    $nlen=pack("N",$len);           // Convert Bytes of len to network order
    if (fwrite($socket,$nlen,4)<4) return 0;  // send length of order to server
    if (fwrite($socket,$order,$len)<$len) return 0;  // send order
    return 1;
}

# Read answer from RRI-Server
#
# Syntax: rriReadData(object socket)
# Parameter:
#    - socket: SSL connection established with rriConnect
#
# Returnvalue:
#    Returns string with answer from RRI server on success or "undef", if an error
#    occured.
sub rriReadData {
        my $sock=shift;
        my ($head, $head2);
        my ($data, $data2);
        my $ret;
        $head="";
        $data="";

        # Step 1: read 4-byte RRI-header
        my $rest=4;
        while ( $rest ) {
            $ret=read $sock,$head2,$rest;
            if (! defined($ret)) {
                   return (undef);
            }
            $head.=$head2;
            $rest-=$ret;
        }
        my $len=unpack "N",$head;
        if ($len > 65535) {        # Should not happen, something went wrong
                   return (undef);
        }

        # Step 2: read payload
        $rest=$len;
        while ( $rest ) {
            $ret=read $sock,$data2,$rest;
            if (! defined($ret)) {
                   return (undef);
            }
            $data.=$data2;
            $rest-=$ret;
        }
        return $data;
}

# Sends order to RRI-Server
#
#
# Syntax: rriSendData(object socket, string order)
# Parameter:
#    - socket: SSL connection established with rriConnect
#    - order: String, which should be send to the RRI-server
#
# Returnvalue:
#    Returns 1 on success, 0 on failure
sub rriSendData {
        my ($sock,$data)=@_;
        my $len=length($data);                # Length of data
        my $head=pack "N",$len;               # convert to 4 byte value in network byteorder
        return 0 if (!print $sock $head);     # send 4 byte header
        return 0 if (!print $sock $data);     # send payload
        return 1;
}

package de.denic.rri.common;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;

import org.apache.commons.lang.Validate;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public final class TcpProtocolFramingHandler implements RriConstants {

  private static final Log log = LogFactory.getLog(TcpProtocolFramingHandler.class);
  // Immutable and thread-safe instance:
  private static final Charset PROTOCOL_CHARACTER_SET = Charset.forName(NAME_OF_PROTOCOLS_CHARACTER_SET);

  // Instance is NOT thread-safe!
  private final CharsetDecoder protocolsCharsetDecoder = PROTOCOL_CHARACTER_SET.newDecoder();
  private final Socket socket;

  public TcpProtocolFramingHandler(final Socket socket) throws IllegalArgumentException {
    super();
    Validate.notNull(socket, "Missing socket");
    this.socket = socket;
  }

  public MessageDataWithOptionalExceptionValue nextFrame() throws IOException {
    return nextFrame(-1); // Keine Prüfung der erlaubten Länge des Frames
  }

  public MessageDataWithOptionalExceptionValue nextFrame(final int maxAllowedPayloadSize) throws IOException {
    return doNextFrame(maxAllowedPayloadSize, socket.getInputStream());
  }

  protected final MessageDataWithOptionalExceptionValue doNextFrame(final int maxAllowedPayloadSize, final InputStream inputStream) throws IOException, IllegalArgumentException {
    final long lengthOfFrameInBytes = evaluateLengthOfFrameInBytes(inputStream);
    if (lengthOfFrameInBytes == 0) {
      return new MessageDataWithOptionalExceptionValue("", null);
    }

    if ((maxAllowedPayloadSize > 0) && (lengthOfFrameInBytes > maxAllowedPayloadSize)) {
      return new MessageDataWithOptionalExceptionValue("[Message to long]", new RriException(RriExceptionType.MESSAGE_TOO_LONG, new Serializable[] {Long.toString(lengthOfFrameInBytes) }), true);
    }

    final int lengthOfFrameInBytesAsInt = (int) lengthOfFrameInBytes;
    final byte[] messageAsByteArray = new byte[lengthOfFrameInBytesAsInt];
    if (log.isInfoEnabled()) {
      log.info("Reading frame of " + lengthOfFrameInBytes + " bytes");
    }
    int bytesReadForMessage = 0;
    while (bytesReadForMessage < lengthOfFrameInBytes) {
        bytesReadForMessage += inputStream.read(messageAsByteArray, bytesReadForMessage, lengthOfFrameInBytesAsInt
								- bytesReadForMessage);
    }
    if (bytesReadForMessage != lengthOfFrameInBytes) {
      throw new IOException("Expecting frame containing minimum of "
								+ lengthOfFrameInBytes + " bytes, but received only " + bytesReadForMessage + " bytes");
    }

    return decodeMessagesBytes(messageAsByteArray);
  }

  private MessageDataWithOptionalExceptionValue decodeMessagesBytes(final byte[] messageAsByteArray)
	throws IllegalArgumentException {
      try {
        final CharBuffer message = protocolsCharsetDecoder.decode(ByteBuffer.wrap(messageAsByteArray));
        return new MessageDataWithOptionalExceptionValue(message.toString(), null);
      }	catch (final CharacterCodingException e) {
        log.warn("Decoding received message failed", e);
        return new MessageDataWithOptionalExceptionValue("[Decoding message data failed]", new RriException(RriExceptionType.MESSAGE_ENCODING_ILLEGAL));
      }
  }

  private long evaluateLengthOfFrameInBytes(final InputStream inputStream) throws IOException {
    long lengthOfFrameInBytes = 0;
    for (int i = 3; i >= 0; i--) {
      final int byteRead = inputStream.read();
      if (byteRead < 0) {
        if (i == 3) {
          // No single byte reaches us
          throw new IOException("Input stream of connection is empty: Connection seems to be closed");
        }

        throw new IOException("Missing four bytes representing frame's length");
      }

      lengthOfFrameInBytes += byteRead << (8 * i);
    }
    if ((lengthOfFrameInBytes > Integer.MAX_VALUE) || (lengthOfFrameInBytes < 0)) {
      throw new IOException("The four bytes representing frame's length denotes a number that cannot be handled: "
        + lengthOfFrameInBytes);
    }

    return lengthOfFrameInBytes;
  }

  public void putFrame(final String frame) throws IOException {
    doPutFrame(frame, socket.getOutputStream());
  }

  /**
  * Access modifier protected for testing purposes
  */
  protected final void doPutFrame(final String frame, final OutputStream outputStream)
  throws UnsupportedEncodingException, IOException {
   final byte[] frameAsUtf8Bytes = frame.getBytes(NAME_OF_PROTOCOLS_CHARACTER_SET);
   final int lengthOfFrameInBytes = frameAsUtf8Bytes.length;
   final byte[] lengthEncodedAsBytes = new byte[] {(byte) (lengthOfFrameInBytes >> 24),
      (byte) (lengthOfFrameInBytes >> 16), (byte) (lengthOfFrameInBytes >> 8), (byte) lengthOfFrameInBytes };
   if (log.isInfoEnabled()) {
     log.info("Writing frame of " + lengthOfFrameInBytes + " bytes");
   }
   synchronized (socket) {
     outputStream.write(lengthEncodedAsBytes);
     outputStream.write(frameAsUtf8Bytes);
     outputStream.flush();
   }
  }

  public void close() {
    if (socket.isClosed()) {
      return;
    }

    try {
      synchronized (socket) {
        try {
           socket.getOutputStream().flush();
        } finally {
          socket.close();
          }
      }
    } catch (final IOException e) {
      log.warn("Closing socket failed", e);
      }
    }

    public InetSocketAddress getSocketAddress() {
    return (InetSocketAddress) socket.getRemoteSocketAddress();
  }

  @Override
  public String toString() {
    return socket.toString();
  }
}