mcs/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 SQLException if a database-access error occurs.
  89          */
  90         public Object FastpathCall(Int32 fnid, Boolean resulttype, FastpathArg[] args)
  91         {
  92             if (conn.BackendProtocolVersion == ProtocolVersion.Version3)
  93             {
  94                 return FastpathV3(fnid, resulttype, args);
  95             }
  96             else
  97             {
  98                 return FastpathV2(fnid, resulttype, args);
  99             }
 100         }
 101
 102         private Object FastpathV3(Int32 fnid, Boolean resulttype, FastpathArg[] args)
 103         {
 104             // give  thread safety
 105             lock (stream)
 106             {
 107                 // send the function call
 108                 try
 109                 {
 110                     Int32 l_msgLen = 0;
 111                     l_msgLen += 16;
 112                     for (Int32 i=0;i < args.Length;i++)
 113                         l_msgLen += args[i].SendSize();
 114
 115                     stream.WriteByte((Byte)'F');
 116                     PGUtil.WriteInt32(stream,l_msgLen);
 117                     PGUtil.WriteInt32(stream,fnid);
 118                     PGUtil.WriteInt16(stream,1);
 119                     PGUtil.WriteInt16(stream,1);
 120                     PGUtil.WriteInt16(stream,(short)args.Length);
 121
 122                     for (Int32 i = 0;i < args.Length;i++)
 123                         args[i].Send(stream);
 124
 125                     PGUtil.WriteInt16(stream,1);
 126
 127                     // This is needed, otherwise data can be lost
 128                     stream.Flush();
 129
 130                 }
 131                 catch (Exception ex)
 132                 {
 133                     throw new Exception(ex.ToString());
 134                 }
 135
 136                 // Now handle the result
 137
 138                 // Now loop, reading the results
 139                 Object result = null; // our result
 140                 Exception error = null;
 141                 Int32 c;
 142                 Boolean l_endQuery = false;
 143                 Byte[] input_buffer = new Byte[512];
 144
 145                 while (!l_endQuery)
 146                 {
 147                     c = (Char)stream.ReadByte();
 148
 149                     switch (c)
 150                     {
 151                     case 'A':   // Asynchronous Notify
 152                         Int32 msglen = PGUtil.ReadInt32(stream,input_buffer);
 153                         Int32 pid = PGUtil.ReadInt32(stream,input_buffer);
 154                         String msg = PGUtil.ReadString(stream,conn.Connector.Encoding);
 155                         PGUtil.ReadString(stream,conn.Connector.Encoding);
 156                         String param = PGUtil.ReadString(stream,conn.Connector.Encoding);
 157
 158                         conn.Connector.CheckErrorsAndNotifications();
 159                         break;
 160                         //------------------------------
 161                         // Error message returned
 162                     case 'E':
 163                         NpgsqlError e = new NpgsqlError(conn.BackendProtocolVersion);
 164                         e.ReadFromStream(stream,conn.Connector.Encoding);
 165                         throw new Exception(e.ToString());
 166
 167                         //------------------------------
 168                         // Notice from backend
 169                     case 'N':
 170                         Int32 l_nlen = PGUtil.ReadInt32(stream,input_buffer);
 171
 172                         NpgsqlError e1 = new NpgsqlError(conn.BackendProtocolVersion);
 173                         e1.ReadFromStream(stream,conn.Connector.Encoding);
 174                         conn.Connector.Mediator.Errors.Add(e1);
 175
 176                         break;
 177
 178                     case 'V':
 179                         Int32 l_msgLen = PGUtil.ReadInt32(stream,input_buffer);
 180                         Int32 l_valueLen = PGUtil.ReadInt32(stream,input_buffer);
 181
 182                         if (l_valueLen == -1)
 183                         {
 184                             //null value
 185                         }
 186                         else if (l_valueLen == 0)
 187                         {
 188                             result = new Byte[0];
 189                         }
 190                         else
 191                         {
 192                             // Return an Integer if
 193                             if (resulttype)
 194
 195                                 result = PGUtil.ReadInt32(stream,input_buffer);
 196                             else
 197                             {
 198                                 Byte[] buf = new Byte[l_valueLen];
 199
 200                                 Int32 bytes_from_stream = 0;
 201                                 Int32 total_bytes_read = 0;
 202                                 Int32 size = l_valueLen;
 203                                 do
 204                                 {
 205                                     bytes_from_stream = stream.Read(buf, total_bytes_read, size);
 206                                     total_bytes_read += bytes_from_stream;
 207                                     size -= bytes_from_stream;
 208                                 }
 209                                 while(size > 0);
 210
 211                                 result = buf;
 212                             }
 213                         }
 214                         break;
 215
 216                     case 'Z':
 217                         //TODO: use size better
 218                         if (PGUtil.ReadInt32(stream,input_buffer) != 5)
 219                             throw new Exception("Received Z" );
 220                         //TODO: handle transaction status
 221                         Char l_tStatus = (Char)stream.ReadByte();
 222                         l_endQuery = true;
 223                         break;
 224
 225                     default:
 226                         throw new Exception("postgresql.fp.protocol received " + c.ToString());
 227                     }
 228                 }
 229
 230                 if ( error != null )
 231                     throw error;
 232
 233                 return result;
 234             }
 235         }
 236
 237         private Object FastpathV2(Int32 fnid, Boolean resulttype, FastpathArg[] args)
 238         {
 239             // added Oct 7 1998 to give us thread safety
 240             lock (stream)
 241             {
 242                 // send the function call
 243                 try
 244                 {
 245                     // 70 is 'F' in ASCII. Note: don't use SendChar() here as it adds padding
 246                     // that confuses the backend. The 0 terminates the command line.
 247                     stream.WriteByte((Byte)70);
 248                     stream.WriteByte((Byte)0);
 249
 250                     PGUtil.WriteInt32(stream,fnid);
 251                     PGUtil.WriteInt32(stream,args.Length);
 252
 253
 254                     for (Int32 i = 0;i < args.Length;i++)
 255                         args[i].Send(stream);
 256
 257                     // This is needed, otherwise data can be lost
 258                     stream.Flush();
 259
 260                 }
 261                 catch (IOException ioe)
 262                 {
 263                     //Should be sending exception as second arg.
 264                     throw new Exception("postgresql.fp.send: "  + ioe.ToString());
 265                 }
 266
 267                 // Now handle the result
 268
 269                 // Now loop, reading the results
 270                 Object result = null; // our result
 271                 String errorMessage = "";
 272                 Byte[] input_buffer = new Byte[512];
 273                 Int32 c;
 274                 Boolean l_endQuery = false;
 275                 while (!l_endQuery)
 276                 {
 277                     c = (Char)stream.ReadByte();
 278
 279                     switch (c)
 280                     {
 281                     case 'A':   // Asynchronous Notify
 282                         //TODO: do something with this
 283                         Int32 pid = PGUtil.ReadInt32(stream,input_buffer);
 284                         String msg = PGUtil.ReadString(stream,conn.Connector.Encoding);
 285
 286
 287                         conn.Connector.CheckErrorsAndNotifications();
 288
 289                         break;
 290
 291                         //------------------------------
 292                         // Error message returned
 293                     case 'E':
 294                         NpgsqlError e = new NpgsqlError(conn.BackendProtocolVersion);
 295                         e.ReadFromStream(stream,conn.Connector.Encoding);
 296                         errorMessage +=  e.Message;
 297                         break;
 298
 299                         //------------------------------
 300                         // Notice from backend
 301                     case 'N':
 302                         NpgsqlError notice = new NpgsqlError(conn.BackendProtocolVersion);
 303                         notice.ReadFromStream(stream,conn.Connector.Encoding);
 304                         errorMessage +=  notice.Message;
 305                         break;
 306
 307                     case 'V':
 308                         Char l_nextChar = (Char)stream.ReadByte();
 309                         if (l_nextChar == 'G')
 310                         {
 311                             Int32 sz = PGUtil.ReadInt32(stream,input_buffer);
 312                             // Return an Integer if
 313                             if (resulttype)
 314                                 result = PGUtil.ReadInt32(stream,input_buffer);
 315                             else
 316                             {
 317                                 Byte[] buf = new Byte[sz];
 318
 319                                 Int32 bytes_from_stream = 0;
 320                                 Int32 total_bytes_read = 0;
 321                                 Int32 size = sz;
 322                                 do
 323                                 {
 324                                     bytes_from_stream = stream.Read(buf, total_bytes_read, size);
 325                                     total_bytes_read += bytes_from_stream;
 326                                     size -= bytes_from_stream;
 327                                 }
 328                                 while(size > 0);
 329
 330                                 result = buf;
 331                             }
 332                             //There should be a trailing '0'
 333                             Int32 l_endChar = (Char)stream.ReadByte();
 334                         }
 335                         else
 336                         {
 337                             //it must have been a '0', thus no results
 338                         }
 339                         break;
 340
 341                     case 'Z':
 342                         l_endQuery = true;
 343                         break;
 344
 345                     default:
 346                         throw new Exception("postgresql.fp.protocol " + c.ToString());
 347                     }
 348                 }
 349
 350                 if ( errorMessage != null )
 351                     throw new Exception("postgresql.fp.error" + errorMessage);
 352
 353                 return result;
 354             }
 355         }
 356
 357         /*
 358          * Send a function call to the PostgreSQL backend by name.
 359          *
 360          * Note: the mapping for the procedure name to function id needs to exist,
 361          * usually to an earlier call to addfunction().
 362          *
 363          * This is the prefered method to call, as function id's can/may change
 364          * between versions of the backend.
 365          *
 366          * For an example of how this works, refer to org.postgresql.largeobject.LargeObject
 367          *
 368          * @param name Function name
 369          * @param resulttype True if the result is an integer, false for other
 370          * results
 371          * @param args FastpathArguments to pass to fastpath
 372          * @return null if no data, Integer if an integer result, or byte[] otherwise
 373          * @exception SQLException if name is unknown or if a database-access error
 374          * occurs.
 375          * @see org.postgresql.largeobject.LargeObject
 376          */
 377         public Object FastpathCall(String name, Boolean resulttype, FastpathArg[] args)
 378         {
 379             return FastpathCall(GetID(name), resulttype, args);
 380         }
 381
 382         /*
 383          * This convenience method assumes that the return value is an Integer
 384          * @param name Function name
 385          * @param args Function arguments
 386          * @return integer result
 387          * @exception SQLException if a database-access error occurs or no result
 388          */
 389         public Int32 GetInteger(String name, FastpathArg[] args)
 390         {
 391             Int32 i = (Int32)FastpathCall(name, true, args);
 392
 393             return i;
 394         }
 395
 396         /*
 397          * This convenience method assumes that the return value is an Integer
 398          * @param name Function name
 399          * @param args Function arguments
 400          * @return byte[] array containing result
 401          * @exception SQLException if a database-access error occurs or no result
 402          */
 403         public Byte[] GetData(String name, FastpathArg[] args)
 404         {
 405             return (Byte[])FastpathCall(name, false, args);
 406         }
 407
 408         /*
 409          * This adds a function to our lookup table.
 410          *
 411          * <p>User code should use the addFunctions method, which is based upon a
 412          * query, rather than hard coding the oid. The oid for a function is not
 413          * guaranteed to remain static, even on different servers of the same
 414          * version.
 415          *
 416          * @param name Function name
 417          * @param fnid Function id
 418          */
 419         public void AddFunction(String name, Int32 fnid)
 420         {
 421             func.Add(name, fnid);
 422         }
 423
 424         /*
 425          * This takes a ResultSet containing two columns. Column 1 contains the
 426          * function name, Column 2 the oid.
 427          *
 428          * <p>It reads the entire ResultSet, loading the values into the function
 429          * table.
 430          *
 431          * <p><b>REMEMBER</b> to close() the resultset after calling this!!
 432          *
 433          * <p><b><em>Implementation note about function name lookups:</em></b>
 434          *
 435          * <p>PostgreSQL stores the function id's and their corresponding names in
 436          * the pg_proc table. To speed things up locally, instead of querying each
 437          * function from that table when required, a Hashtable is used. Also, only
 438          * the function's required are entered into this table, keeping connection
 439          * times as fast as possible.
 440          *
 441          * <p>The org.postgresql.largeobject.LargeObject class performs a query upon it's startup,
 442          * and passes the returned ResultSet to the addFunctions() method here.
 443          *
 444          * <p>Once this has been done, the LargeObject api refers to the functions by
 445          * name.
 446          *
 447          * <p>Dont think that manually converting them to the oid's will work. Ok,
 448          * they will for now, but they can change during development (there was some
 449          * discussion about this for V7.0), so this is implemented to prevent any
 450          * unwarranted headaches in the future.
 451          *
 452          * @param rs ResultSet
 453          * @exception SQLException if a database-access error occurs.
 454          * @see org.postgresql.largeobject.LargeObjectManager
 455          */
 456         public void AddFunctions(IDataReader rs)
 457         {
 458             while (rs.Read())
 459             {
 460                 String key = (String)rs[0];
 461                 if( !func.ContainsKey(key ) )
 462                     func.Add(key, Int32.Parse(rs[1].ToString()));
 463             }
 464         }
 465
 466         /*
 467          * This returns the function id associated by its name
 468          *
 469          * <p>If addFunction() or addFunctions() have not been called for this name,
 470          * then an SQLException is thrown.
 471          *
 472          * @param name Function name to lookup
 473          * @return Function ID for fastpath call
 474          * @exception SQLException is function is unknown.
 475          */
 476         public Int32 GetID(String name)
 477         {
 478             Int32 id = (Int32)func[name];
 479
 480
 481
 482             return id;
 483         }
 484     }
 485
 486 }