class/Npgsql/Npgsql/NpgsqlCopyOutStream.cs

   1 // Npgsql.NpgsqlCopyOutStream.cs
   2 //
   3 // Author:
   4 //     Kalle Hallivuori <kato@iki.fi>
   5 //
   6 //    Copyright (C) 2007 The Npgsql Development Team
   7 //    npgsql-general@gborg.postgresql.org
   8 //    http://gborg.postgresql.org/project/npgsql/projdisplay.php
   9 //
  10 // Permission to use, copy, modify, and distribute this software and its
  11 // documentation for any purpose, without fee, and without a written
  12 // agreement is hereby granted, provided that the above copyright notice
  13 // and this paragraph and the following two paragraphs appear in all copies.
  14 //
  15 // IN NO EVENT SHALL THE NPGSQL DEVELOPMENT TEAM BE LIABLE TO ANY PARTY
  16 // FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES,
  17 // INCLUDING LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
  18 // DOCUMENTATION, EVEN IF THE NPGSQL DEVELOPMENT TEAM HAS BEEN ADVISED OF
  19 // THE POSSIBILITY OF SUCH DAMAGE.
  20 //
  21 // THE NPGSQL DEVELOPMENT TEAM SPECIFICALLY DISCLAIMS ANY WARRANTIES,
  22 // INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
  23 // AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS
  24 // ON AN "AS IS" BASIS, AND THE NPGSQL DEVELOPMENT TEAM HAS NO OBLIGATIONS
  25 // TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
  26
  27
  28 using System;
  29 using System.IO;
  30
  31 namespace Npgsql
  32 {
  33         /// <summary>
  34         /// Stream for reading data from a table or select on a PostgreSQL version 7.4 or newer database during an active COPY TO STDOUT operation.
  35         /// <b>Passes data exactly as provided by the server.</b>
  36         /// </summary>
  37         internal class NpgsqlCopyOutStream : Stream
  38         {
  39                 private NpgsqlConnector _context;
  40                 private long _bytesPassed = 0;
  41                 private byte[] _buf = null;
  42                 private int _bufOffset = 0;
  43
  44                 /// <summary>
  45                 /// True while this stream can be used to read copy data from server
  46                 /// </summary>
  47                 private bool IsActive
  48                 {
  49                         get { return _context != null && _context.CurrentState is NpgsqlCopyOutState && _context.Mediator.CopyStream == this; }
  50                 }
  51
  52                 /// <summary>
  53                 /// Created only by NpgsqlCopyOutState.StartCopy()
  54                 /// </summary>
  55                 internal NpgsqlCopyOutStream(NpgsqlConnector context)
  56                 {
  57                         _context = context;
  58                 }
  59
  60                 /// <summary>
  61                 /// True
  62                 /// </summary>
  63                 public override bool CanRead
  64                 {
  65                         get { return true; }
  66                 }
  67
  68                 /// <summary>
  69                 /// False
  70                 /// </summary>
  71                 public override bool CanWrite
  72                 {
  73                         get { return false; }
  74                 }
  75
  76                 /// <summary>
  77                 /// False
  78                 /// </summary>
  79                 public override bool CanSeek
  80                 {
  81                         get { return false; }
  82                 }
  83
  84                 /// <summary>
  85                 /// Number of bytes read so far
  86                 /// </summary>
  87                 public override long Length
  88                 {
  89                         get { return _bytesPassed; }
  90                 }
  91
  92                 /// <summary>
  93                 /// Number of bytes read so far; can not be set.
  94                 /// </summary>
  95                 public override long Position
  96                 {
  97                         get { return _bytesPassed; }
  98                         set { throw new NotSupportedException("Tried to set Position of network stream " + this); }
  99                 }
 100
 101                 /// <summary>
 102                 /// Discards copy data as long as server pushes it. Returns after operation is finished.
 103                 /// Does nothing if this stream is not the active copy operation reader.
 104                 /// </summary>
 105                 public override void Close()
 106                 {
 107                         if (_context != null)
 108                         {
 109                                 if (IsActive)
 110                                 {
 111                                         while (_context.CurrentState.GetCopyData(_context) != null)
 112                                         {
 113                                                 ; // flush rest
 114                                         }
 115                                 }
 116                                 if (_context.Mediator.CopyStream == this)
 117                                 {
 118                                         _context.Mediator.CopyStream = null;
 119                                 }
 120                                 _context = null;
 121                         }
 122                 }
 123
 124                 /// <summary>
 125                 /// Not writable.
 126                 /// </summary>
 127                 public override void Write(byte[] buf, int off, int len)
 128                 {
 129                         throw new NotSupportedException("Tried to write non-writable " + this);
 130                 }
 131
 132                 /// <summary>
 133                 /// Not flushable.
 134                 /// </summary>
 135                 public override void Flush()
 136                 {
 137                         throw new NotSupportedException("Tried to flush read-only " + this);
 138                 }
 139
 140                 /// <summary>
 141                 /// Copies data read from server to given byte buffer.
 142                 /// Since server returns data row by row, length will differ each time, but it is only zero once the operation ends.
 143                 /// Can be mixed with calls to the more efficient NpgsqlCopyOutStream.Read() : byte[] though that would not make much sense.
 144                 /// </summary>
 145                 public override int Read(byte[] buf, int off, int len)
 146                 {
 147                         if (! IsActive)
 148                         {
 149                                 throw new ObjectDisposedException("Reading from closed " + this);
 150                         }
 151
 152                         if (_buf == null) // otherwise _buf still contains data that did not fit into request buffer in an earlier call
 153                         {
 154                                 _buf = Read();
 155                                 _bufOffset = 0;
 156                         }
 157                         if (off + len > buf.Length)
 158                         {
 159                                 len = buf.Length - off;
 160                         }
 161
 162                         int i = 0;
 163                         if (_buf != null)
 164                         {
 165                                 for (; _bufOffset < _buf.Length && i < len; i++)
 166                                 {
 167                                         buf[off + i] = _buf[_bufOffset++];
 168                                 }
 169                                 if (_bufOffset >= _buf.Length)
 170                                 {
 171                                         _buf = null; // whole of our contents fit into request buffer
 172                                 }
 173                                 _bytesPassed += i;
 174                         }
 175                         return i;
 176                 }
 177
 178                 /// <summary>
 179                 /// Not seekable
 180                 /// </summary>
 181                 public override long Seek(long pos, SeekOrigin so)
 182                 {
 183                         throw new NotSupportedException("Tried to seek non-seekable " + this);
 184                 }
 185
 186                 /// <summary>
 187                 /// Not supported
 188                 /// </summary>
 189                 public override void SetLength(long len)
 190                 {
 191                         throw new NotSupportedException("Tried to set length of network stream " + this);
 192                 }
 193
 194                 /// <summary>
 195                 /// Returns a whole row of data from server without extra work.
 196                 /// If standard Stream.Read(...) has been called before, it's internal buffers remains are returned.
 197                 /// </summary>
 198                 public byte[] Read()
 199                 {
 200                         byte[] result;
 201                         if (_buf == null)
 202                         {
 203                                 result = _context.CurrentState.GetCopyData(_context);
 204                         }
 205                         else if (_bufOffset < 1)
 206                         {
 207                                 result = _buf;
 208                         }
 209                         else
 210                         {
 211                                 result = new byte[_buf.Length - _bufOffset];
 212                                 for (int i = 0; i < result.Length; i++)
 213                                 {
 214                                         result[i] = _buf[_bufOffset + i];
 215                                 }
 216                                 _buf = null;
 217                         }
 218                         return result;
 219                 }
 220         }
 221 }