postgres/src/interfaces/jdbc/org/postgresql/PG_Stream.java

package org.postgresql;

import java.io.*;
import java.lang.*;
import java.net.*;
import java.util.*;
import java.sql.*;
import java.security.*;
import org.postgresql.*;
import org.postgresql.core.*;
import org.postgresql.util.*;

/**
 * $Id: PG_Stream.java,v 1.12 2001/08/26 01:06:20 momjian Exp $
 *
 * This class is used by Connection & PGlobj for communicating with the
 * backend.
 *
 * @see java.sql.Connection
 */
//  This class handles all the Streamed I/O for a org.postgresql connection
public class PG_Stream
{
  private Socket connection;
  private InputStream pg_input;
  private BufferedOutputStream pg_output;
  private byte[] byte_buf = new byte[8*1024];

    BytePoolDim1 bytePoolDim1 = new BytePoolDim1();
    BytePoolDim2 bytePoolDim2 = new BytePoolDim2();

   private static class PrivilegedSocket
      implements PrivilegedExceptionAction
   {
      private String host;
      private int port;

      PrivilegedSocket(String host, int port)
      {
         this.host = host;
         this.port = port;
      }

      public Object run() throws Exception
      {
         return new Socket(host, port);
      }
   }


  /**
   * Constructor:  Connect to the PostgreSQL back end and return
   * a stream connection.
   *
   * @param host the hostname to connect to
   * @param port the port number that the postmaster is sitting on
   * @exception IOException if an IOException occurs below it.
   */
  public PG_Stream(String host, int port) throws IOException
  {
     PrivilegedSocket ps = new PrivilegedSocket(host, port);
     try {
        connection = (Socket)AccessController.doPrivileged(ps);
     }
     catch(PrivilegedActionException pae){
        throw (IOException)pae.getException();
     }

    // Submitted by Jason Venner <jason@idiom.com> adds a 10x speed
    // improvement on FreeBSD machines (caused by a bug in their TCP Stack)
    connection.setTcpNoDelay(true);

    // Buffer sizes submitted by Sverre H Huseby <sverrehu@online.no>
    pg_input = new BufferedInputStream(connection.getInputStream(), 8192);
    pg_output = new BufferedOutputStream(connection.getOutputStream(), 8192);
  }

  /**
   * Sends a single character to the back end
   *
   * @param val the character to be sent
   * @exception IOException if an I/O error occurs
   */
  public void SendChar(int val) throws IOException
  {
      pg_output.write((byte)val);
  }

  /**
   * Sends an integer to the back end
   *
   * @param val the integer to be sent
   * @param siz the length of the integer in bytes (size of structure)
   * @exception IOException if an I/O error occurs
   */
  public void SendInteger(int val, int siz) throws IOException
  {
    byte[] buf = bytePoolDim1.allocByte(siz);

    while (siz-- > 0)
      {
	buf[siz] = (byte)(val & 0xff);
	val >>= 8;
      }
    Send(buf);
  }

  /**
   * Send an array of bytes to the backend
   *
   * @param buf The array of bytes to be sent
   * @exception IOException if an I/O error occurs
   */
  public void Send(byte buf[]) throws IOException
  {
    pg_output.write(buf);
  }

  /**
   * Send an exact array of bytes to the backend - if the length
   * has not been reached, send nulls until it has.
   *
   * @param buf the array of bytes to be sent
   * @param siz the number of bytes to be sent
   * @exception IOException if an I/O error occurs
   */
  public void Send(byte buf[], int siz) throws IOException
  {
    Send(buf,0,siz);
  }

  /**
   * Send an exact array of bytes to the backend - if the length
   * has not been reached, send nulls until it has.
   *
   * @param buf the array of bytes to be sent
   * @param off offset in the array to start sending from
   * @param siz the number of bytes to be sent
   * @exception IOException if an I/O error occurs
   */
  public void Send(byte buf[], int off, int siz) throws IOException
  {
    int i;

    pg_output.write(buf, off, ((buf.length-off) < siz ? (buf.length-off) : siz));
    if((buf.length-off) < siz)
      {
	for (i = buf.length-off ; i < siz ; ++i)
	  {
	    pg_output.write(0);
	  }
      }
  }

  /**
   * Receives a single character from the backend
   *
   * @return the character received
   * @exception SQLException if an I/O Error returns
   */
  public int ReceiveChar() throws SQLException
  {
    int c = 0;

    try
      {
	c = pg_input.read();
	if (c < 0) throw new PSQLException("postgresql.stream.eof");
      } catch (IOException e) {
	throw new PSQLException("postgresql.stream.ioerror",e);
      }
      return c;
  }

  /**
   * Receives an integer from the backend
   *
   * @param siz length of the integer in bytes
   * @return the integer received from the backend
   * @exception SQLException if an I/O error occurs
   */
  public int ReceiveInteger(int siz) throws SQLException
  {
    int n = 0;

    try
      {
	for (int i = 0 ; i < siz ; i++)
	  {
	    int b = pg_input.read();

	    if (b < 0)
	      throw new PSQLException("postgresql.stream.eof");
	    n = n | (b << (8 * i)) ;
	  }
      } catch (IOException e) {
	throw new PSQLException("postgresql.stream.ioerror",e);
      }
      return n;
  }

  /**
   * Receives an integer from the backend
   *
   * @param siz length of the integer in bytes
   * @return the integer received from the backend
   * @exception SQLException if an I/O error occurs
   */
  public int ReceiveIntegerR(int siz) throws SQLException
  {
    int n = 0;

    try
      {
	for (int i = 0 ; i < siz ; i++)
	  {
	    int b = pg_input.read();

	    if (b < 0)
	      throw new PSQLException("postgresql.stream.eof");
	    n = b | (n << 8);
	  }
      } catch (IOException e) {
	throw new PSQLException("postgresql.stream.ioerror",e);
      }
      return n;
  }

  /**
   * Receives a null-terminated string from the backend.  If we don't see a
   * null, then we assume something has gone wrong.
   *
   * @param encoding the charset encoding to use.
   * @return string from back end
   * @exception SQLException if an I/O error occurs, or end of file
   */
  public String ReceiveString(Encoding encoding)
      throws SQLException
  {
    int s = 0;
    byte[] rst = byte_buf;
    try {
      int buflen = rst.length;
      boolean done = false;
      while (!done) {
        while (s < buflen) {
	    int c = pg_input.read();
	    if (c < 0)
	      throw new PSQLException("postgresql.stream.eof");
 	    else if (c == 0) {
 		rst[s] = 0;
            done = true;
 		break;
          } else {
	      rst[s++] = (byte)c;
	  }
          if (s >= buflen) { // Grow the buffer
            buflen = (int)(buflen*2); // 100% bigger
            byte[] newrst = new byte[buflen];
            System.arraycopy(rst, 0, newrst, 0, s);
            rst = newrst;
          }
        }
      }
      } catch (IOException e) {
	throw new PSQLException("postgresql.stream.ioerror",e);
      }
      return encoding.decode(rst, 0, s);
  }

  /**
   * Read a tuple from the back end.  A tuple is a two dimensional
   * array of bytes
   *
   * @param nf the number of fields expected
   * @param bin true if the tuple is a binary tuple
   * @return null if the current response has no more tuples, otherwise
   *	an array of strings
   * @exception SQLException if a data I/O error occurs
   */
  public byte[][] ReceiveTuple(int nf, boolean bin) throws SQLException
  {
    int i, bim = (nf + 7)/8;
    byte[] bitmask = Receive(bim);
    byte[][] answer = bytePoolDim2.allocByte(nf);

    int whichbit = 0x80;
    int whichbyte = 0;

    for (i = 0 ; i < nf ; ++i)
      {
	boolean isNull = ((bitmask[whichbyte] & whichbit) == 0);
	whichbit >>= 1;
	if (whichbit == 0)
	  {
	    ++whichbyte;
	    whichbit = 0x80;
	  }
	if (isNull)
	  answer[i] = null;
	else
	  {
	    int len = ReceiveIntegerR(4);
	    if (!bin)
	      len -= 4;
	    if (len < 0)
	      len = 0;
	    answer[i] = Receive(len);
	  }
      }
    return answer;
  }

  /**
   * Reads in a given number of bytes from the backend
   *
   * @param siz number of bytes to read
   * @return array of bytes received
   * @exception SQLException if a data I/O error occurs
   */
  private byte[] Receive(int siz) throws SQLException
  {
      byte[] answer = bytePoolDim1.allocByte(siz);
    Receive(answer,0,siz);
    return answer;
  }

  /**
   * Reads in a given number of bytes from the backend
   *
   * @param buf buffer to store result
   * @param off offset in buffer
   * @param siz number of bytes to read
   * @exception SQLException if a data I/O error occurs
   */
  public void Receive(byte[] b,int off,int siz) throws SQLException
  {
    int s = 0;

    try
      {
	while (s < siz)
	  {
	    int w = pg_input.read(b, off+s, siz - s);
	    if (w < 0)
	      throw new PSQLException("postgresql.stream.eof");
	    s += w;
	  }
      } catch (IOException e) {
	  throw new PSQLException("postgresql.stream.ioerror",e);
      }
  }

  /**
   * This flushes any pending output to the backend. It is used primarily
   * by the Fastpath code.
   * @exception SQLException if an I/O error occurs
   */
  public void flush() throws SQLException
  {
    try {
      pg_output.flush();
    } catch (IOException e) {
      throw new PSQLException("postgresql.stream.flush",e);
    }
  }

  /**
   * Closes the connection
   *
   * @exception IOException if a IO Error occurs
   */
  public void close() throws IOException
  {
    pg_output.close();
    pg_input.close();
    connection.close();
  }

}