class/Npgsql/NpgsqlTypes/FastPath.cs

   1 /*-------------------------------------------------------------------------
   2
   3   Fastpath.cs
   4       This class is a port of the class Fastpath.java implemented by
   5       PostgreSQL Global Development Group
   6
   7  Copyright (c) 2004, Emiliano Necciari
   8  Original Code: Copyright (c) 2003, PostgreSQL Global Development Group
   9
  10  Note: (Francisco Figueiredo Jr.)
  11         Changed case of method names to conform to .Net names standard.
  12         Also changed type names to their true names. i.e. int -> Int32
  13
  14  This library is free software; you can redistribute it and/or
  15  modify it under the terms of the GNU Lesser General Public
  16  License as published by the Free Software Foundation; either
  17  version 2.1 of the License, or (at your option) any later version.
  18
  19  This library is distributed in the hope that it will be useful,
  20  but WITHOUT ANY WARRANTY; without even the implied warranty of
  21  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  22  Lesser General Public License for more details.
  23
  24  You should have received a copy of the GNU Lesser General Public
  25  License along with this library; if not, write to the Free Software
  26  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  27 -------------------------------------------------------------------------
  28 */
  29
  30 using System;
  31 using System.Collections;
  32 using System.IO;
  33 using System.Data;
  34 using Npgsql;
  35
  36 namespace NpgsqlTypes
  37 {
  38
  39
  40     /*
  41      * This class implements the Fastpath api.
  42      *
  43      *
  44      *
  45      */
  46     public class Fastpath
  47     {
  48         // This maps the functions names to their id's (possible unique just
  49         // to a connection).
  50         protected Hashtable func = new Hashtable();
  51
  52         protected NpgsqlConnection conn;                // our connection
  53         protected Stream  stream;       // the network stream
  54
  55         /*
  56          * Initialises the fastpath system
  57          *
  58          * @param conn BaseConnection to attach to
  59          * @param stream The network stream to the backend
  60          */
  61         public Fastpath(NpgsqlConnection conn, Stream stream)
  62         {
  63             this.conn = conn;
  64             this.stream = stream;
  65         }
  66
  67
  68         /*
  69          * Initialises the fastpath system
  70          *
  71          * @param conn BaseConnection to attach to
  72          * @param stream The network stream to the backend
  73          */
  74         public Fastpath(NpgsqlConnection conn)
  75         {
  76             this.conn = conn;
  77             // check if the connection is closed ?
  78             this.stream = conn.Connector.Stream;
  79         }
  80
  81         /*
  82          * Send a function call to the PostgreSQL backend
  83          *
  84          * @param fnid Function id
  85          * @param resulttype True if the result is an integer, false for other results
  86          * @param args FastpathArguments to pass to fastpath
  87          * @return null if no data, Integer if an integer result, or byte[] otherwise
  88          * @exception NpgsqlException if a database-access error occurs.
  89          */
  90         public Object FastpathCall(Int32 fnid, Boolean resulttype, FastpathArg[] args)
  91         {
  92             try
  93             {
  94                 if (conn.BackendProtocolVersion == ProtocolVersion.Version3)
  95                 {
  96                     return FastpathV3(fnid, resulttype, args);
  97                 }
  98                 else
  99                 {
 100                     return FastpathV2(fnid, resulttype, args);
 101                 }
 102             }
 103             catch(IOException e)
 104             {
 105                 conn.ClearPool();
 106                 throw new NpgsqlException("The Connection is broken.");
 107             }
 108         }
 109
 110         private Object FastpathV3(Int32 fnid, Boolean resulttype, FastpathArg[] args)
 111         {
 112             // give  thread safety
 113             lock (stream)
 114             {
 115                 // send the function call
 116
 117                 {
 118                     Int32 l_msgLen = 0;
 119                     l_msgLen += 16;
 120                     for (Int32 i=0;i < args.Length;i++)
 121                         l_msgLen += args[i].SendSize();
 122
 123                     stream.WriteByte((Byte)'F');
 124                     PGUtil.WriteInt32(stream,l_msgLen);
 125                     PGUtil.WriteInt32(stream,fnid);
 126                     PGUtil.WriteInt16(stream,1);
 127                     PGUtil.WriteInt16(stream,1);
 128                     PGUtil.WriteInt16(stream,(short)args.Length);
 129
 130                     for (Int32 i = 0;i < args.Length;i++)
 131                         args[i].Send(stream);
 132
 133                     PGUtil.WriteInt16(stream,1);
 134
 135                     // This is needed, otherwise data can be lost
 136                     stream.Flush();
 137                 }
 138
 139
 140                 // Now handle the result
 141
 142                 // Now loop, reading the results
 143                 Object result = null; // our result
 144                 Exception error = null;
 145                 Int32 c;
 146                 Boolean l_endQuery = false;
 147                 Byte[] input_buffer = new Byte[512];
 148
 149                 while (!l_endQuery)
 150                 {
 151                     c = (Char)stream.ReadByte();
 152
 153                     switch (c)
 154                     {
 155                     case 'A':   // Asynchronous Notify
 156                         Int32 msglen = PGUtil.ReadInt32(stream,input_buffer);
 157                         Int32 pid = PGUtil.ReadInt32(stream,input_buffer);
 158                         String msg = PGUtil.ReadString(stream,conn.Connector.Encoding);
 159                         PGUtil.ReadString(stream,conn.Connector.Encoding);
 160                         String param = PGUtil.ReadString(stream,conn.Connector.Encoding);
 161
 162                         conn.Connector.CheckErrorsAndNotifications();
 163                         break;
 164                         //------------------------------
 165                         // Error message returned
 166                     case 'E':
 167                         NpgsqlError e = new NpgsqlError(conn.BackendProtocolVersion);
 168                         e.ReadFromStream(stream,conn.Connector.Encoding);
 169                         throw new NpgsqlException(e.ToString());
 170
 171                         //------------------------------
 172                         // Notice from backend
 173                     case 'N':
 174                         Int32 l_nlen = PGUtil.ReadInt32(stream,input_buffer);
 175
 176                         NpgsqlError e1 = new NpgsqlError(conn.BackendProtocolVersion);
 177                         e1.ReadFromStream(stream,conn.Connector.Encoding);
 178                         conn.Connector.Mediator.Errors.Add(e1);
 179
 180                         break;
 181
 182                     case 'V':
 183                         Int32 l_msgLen = PGUtil.ReadInt32(stream,input_buffer);
 184                         Int32 l_valueLen = PGUtil.ReadInt32(stream,input_buffer);
 185
 186                         if (l_valueLen == -1)
 187                         {
 188                             //null value
 189                         }
 190                         else if (l_valueLen == 0)
 191                         {
 192                             result = new Byte[0];
 193                         }
 194                         else
 195                         {
 196                             // Return an Integer if
 197                             if (resulttype)
 198
 199                                 result = PGUtil.ReadInt32(stream,input_buffer);
 200                             else
 201                             {
 202                                 Byte[] buf = new Byte[l_valueLen];
 203
 204                                 Int32 bytes_from_stream = 0;
 205                                 Int32 total_bytes_read = 0;
 206                                 Int32 size = l_valueLen;
 207                                 do
 208                                 {
 209                                     bytes_from_stream = stream.Read(buf, total_bytes_read, size);
 210                                     total_bytes_read += bytes_from_stream;
 211                                     size -= bytes_from_stream;
 212                                 }
 213                                 while(size > 0);
 214
 215                                 result = buf;
 216                             }
 217                         }
 218                         break;
 219
 220                     case 'Z':
 221                         //TODO: use size better
 222                         if (PGUtil.ReadInt32(stream,input_buffer) != 5)
 223                             throw new NpgsqlException("Received Z" );
 224                         //TODO: handle transaction status
 225                         Char l_tStatus = (Char)stream.ReadByte();
 226                         l_endQuery = true;
 227                         break;
 228
 229                     default:
 230                         throw new NpgsqlException("postgresql.fp.protocol received " + c.ToString());
 231                     }
 232                 }
 233
 234                 if ( error != null )
 235                     throw error;
 236
 237                 return result;
 238             }
 239         }
 240
 241         private Object FastpathV2(Int32 fnid, Boolean resulttype, FastpathArg[] args)
 242         {
 243             // added Oct 7 1998 to give us thread safety
 244             lock (stream)
 245             {
 246                 // send the function call
 247
 248                 // 70 is 'F' in ASCII. Note: don't use SendChar() here as it adds padding
 249                 // that confuses the backend. The 0 terminates the command line.
 250                 stream.WriteByte((Byte)70);
 251                 stream.WriteByte((Byte)0);
 252
 253                 PGUtil.WriteInt32(stream,fnid);
 254                 PGUtil.WriteInt32(stream,args.Length);
 255
 256
 257                 for (Int32 i = 0;i < args.Length;i++)
 258                     args[i].Send(stream);
 259
 260                 // This is needed, otherwise data can be lost
 261                 stream.Flush();
 262
 263
 264                 // Now handle the result
 265
 266                 // Now loop, reading the results
 267                 Object result = null; // our result
 268                 String errorMessage = "";
 269                 Byte[] input_buffer = new Byte[512];
 270                 Int32 c;
 271                 Boolean l_endQuery = false;
 272                 while (!l_endQuery)
 273                 {
 274                     c = (Char)stream.ReadByte();
 275
 276                     switch (c)
 277                     {
 278                     case 'A':   // Asynchronous Notify
 279                         //TODO: do something with this
 280                         Int32 pid = PGUtil.ReadInt32(stream,input_buffer);
 281                         String msg = PGUtil.ReadString(stream,conn.Connector.Encoding);
 282
 283
 284                         conn.Connector.CheckErrorsAndNotifications();
 285
 286                         break;
 287
 288                         //------------------------------
 289                         // Error message returned
 290                     case 'E':
 291                         NpgsqlError e = new NpgsqlError(conn.BackendProtocolVersion);
 292                         e.ReadFromStream(stream,conn.Connector.Encoding);
 293                         errorMessage +=  e.Message;
 294                         break;
 295
 296                         //------------------------------
 297                         // Notice from backend
 298                     case 'N':
 299                         NpgsqlError notice = new NpgsqlError(conn.BackendProtocolVersion);
 300                         notice.ReadFromStream(stream,conn.Connector.Encoding);
 301                         errorMessage +=  notice.Message;
 302                         break;
 303
 304                     case 'V':
 305                         Char l_nextChar = (Char)stream.ReadByte();
 306                         if (l_nextChar == 'G')
 307                         {
 308                             Int32 sz = PGUtil.ReadInt32(stream,input_buffer);
 309                             // Return an Integer if
 310                             if (resulttype)
 311                                 result = PGUtil.ReadInt32(stream,input_buffer);
 312                             else
 313                             {
 314                                 Byte[] buf = new Byte[sz];
 315
 316                                 Int32 bytes_from_stream = 0;
 317                                 Int32 total_bytes_read = 0;
 318                                 Int32 size = sz;
 319                                 do
 320                                 {
 321                                     bytes_from_stream = stream.Read(buf, total_bytes_read, size);
 322                                     total_bytes_read += bytes_from_stream;
 323                                     size -= bytes_from_stream;
 324                                 }
 325                                 while(size > 0);
 326
 327                                 result = buf;
 328                             }
 329                             //There should be a trailing '0'
 330                             Int32 l_endChar = (Char)stream.ReadByte();
 331                         }
 332                         else
 333                         {
 334                             //it must have been a '0', thus no results
 335                         }
 336                         break;
 337
 338                     case 'Z':
 339                         l_endQuery = true;
 340                         break;
 341
 342                     default:
 343                         throw new NpgsqlException("postgresql.fp.protocol " + c.ToString());
 344                     }
 345                 }
 346
 347                 if ( errorMessage != null )
 348                     throw new NpgsqlException("postgresql.fp.error" + errorMessage);
 349
 350                 return result;
 351             }
 352         }
 353
 354         /*
 355          * Send a function call to the PostgreSQL backend by name.
 356          *
 357          * Note: the mapping for the procedure name to function id needs to exist,
 358          * usually to an earlier call to addfunction().
 359          *
 360          * This is the prefered method to call, as function id's can/may change
 361          * between versions of the backend.
 362          *
 363          * For an example of how this works, refer to org.postgresql.largeobject.LargeObject
 364          *
 365          * @param name Function name
 366          * @param resulttype True if the result is an integer, false for other
 367          * results
 368          * @param args FastpathArguments to pass to fastpath
 369          * @return null if no data, Integer if an integer result, or byte[] otherwise
 370          * @exception NpgsqlException if name is unknown or if a database-access error
 371          * occurs.
 372          * @see org.postgresql.largeobject.LargeObject
 373          */
 374         public Object FastpathCall(String name, Boolean resulttype, FastpathArg[] args)
 375         {
 376             return FastpathCall(GetID(name), resulttype, args);
 377         }
 378
 379         /*
 380          * This convenience method assumes that the return value is an Integer
 381          * @param name Function name
 382          * @param args Function arguments
 383          * @return integer result
 384          * @exception NpgsqlException if a database-access error occurs or no result
 385          */
 386         public Int32 GetInteger(String name, FastpathArg[] args)
 387         {
 388             Int32 i = (Int32)FastpathCall(name, true, args);
 389
 390             return i;
 391         }
 392
 393         /*
 394          * This convenience method assumes that the return value is an Integer
 395          * @param name Function name
 396          * @param args Function arguments
 397          * @return byte[] array containing result
 398          * @exception NpgsqlException if a database-access error occurs or no result
 399          */
 400         public Byte[] GetData(String name, FastpathArg[] args)
 401         {
 402             return (Byte[])FastpathCall(name, false, args);
 403         }
 404
 405         /*
 406          * This adds a function to our lookup table.
 407          *
 408          * <p>User code should use the addFunctions method, which is based upon a
 409          * query, rather than hard coding the oid. The oid for a function is not
 410          * guaranteed to remain static, even on different servers of the same
 411          * version.
 412          *
 413          * @param name Function name
 414          * @param fnid Function id
 415          */
 416         public void AddFunction(String name, Int32 fnid)
 417         {
 418             func.Add(name, fnid);
 419         }
 420
 421         /*
 422          * This takes a ResultSet containing two columns. Column 1 contains the
 423          * function name, Column 2 the oid.
 424          *
 425          * <p>It reads the entire ResultSet, loading the values into the function
 426          * table.
 427          *
 428          * <p><b>REMEMBER</b> to close() the resultset after calling this!!
 429          *
 430          * <p><b><em>Implementation note about function name lookups:</em></b>
 431          *
 432          * <p>PostgreSQL stores the function id's and their corresponding names in
 433          * the pg_proc table. To speed things up locally, instead of querying each
 434          * function from that table when required, a Hashtable is used. Also, only
 435          * the function's required are entered into this table, keeping connection
 436          * times as fast as possible.
 437          *
 438          * <p>The org.postgresql.largeobject.LargeObject class performs a query upon it's startup,
 439          * and passes the returned ResultSet to the addFunctions() method here.
 440          *
 441          * <p>Once this has been done, the LargeObject api refers to the functions by
 442          * name.
 443          *
 444          * <p>Dont think that manually converting them to the oid's will work. Ok,
 445          * they will for now, but they can change during development (there was some
 446          * discussion about this for V7.0), so this is implemented to prevent any
 447          * unwarranted headaches in the future.
 448          *
 449          * @param rs ResultSet
 450          * @exception NpgsqlException if a database-access error occurs.
 451          * @see org.postgresql.largeobject.LargeObjectManager
 452          */
 453         public void AddFunctions(IDataReader rs)
 454         {
 455             while (rs.Read())
 456             {
 457                 String key = (String)rs[0];
 458                 if( !func.ContainsKey(key ) )
 459                     func.Add(key, Int32.Parse(rs[1].ToString()));
 460             }
 461         }
 462
 463         /*
 464          * This returns the function id associated by its name
 465          *
 466          * <p>If addFunction() or addFunctions() have not been called for this name,
 467          * then an NpgsqlException is thrown.
 468          *
 469          * @param name Function name to lookup
 470          * @return Function ID for fastpath call
 471          * @exception NpgsqlException is function is unknown.
 472          */
 473         public Int32 GetID(String name)
 474         {
 475             Int32 id = (Int32)func[name];
 476
 477
 478
 479             return id;
 480         }
 481     }
 482
 483 }