GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — Npgsql/src/Npgsql/NpgsqlLargeObjectStream.cs at main · roji/Npgsql Expand file tree/ Copy pathNpgsqlLargeObjectStream.csMore file actions 318 lines (277 loc) · 14.1 KB / Copy pathNpgsqlLargeObjectStream.csFile metadata and controls318 lines (277 loc) · 14.1 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 using System; using System.IO; using System.Threading; using System.Threading.Tasks; namespace Npgsql { /// <summary> /// An interface to remotely control the seekable stream for an opened large object on a PostgreSQL server. /// Note that the OpenRead/OpenReadWrite method as well as all operations performed on this stream must be wrapped inside a database transaction. /// </summary> public sealed class NpgsqlLargeObjectStream : Stream { readonly NpgsqlLargeObjectManager _manager; readonly int _fd; long _pos; readonly bool _writeable; bool _disposed; internal NpgsqlLargeObjectStream(NpgsqlLargeObjectManager manager, int fd, bool writeable) { _manager = manager; _fd = fd; _pos = 0; _writeable = writeable; } void CheckDisposed() { if (_disposed) throw new InvalidOperationException("Object disposed"); } /// <summary> /// Since PostgreSQL 9.3, large objects larger than 2GB can be handled, up to 4TB. /// This property returns true whether the PostgreSQL version is >= 9.3. /// </summary> public bool Has64BitSupport => _manager.Connection.PostgreSqlVersion >= new Version(9, 3); /// <summary> /// Reads <i>count</i> bytes from the large object. The only case when fewer bytes are read is when end of stream is reached. /// </summary> /// <param name="buffer">The buffer where read data should be stored.</param> /// <param name="offset">The offset in the buffer where the first byte should be read.</param> /// <param name="count">The maximum number of bytes that should be read.</param> /// <returns>How many bytes actually read, or 0 if end of file was already reached.</returns> public override int Read(byte[] buffer, int offset, int count) => Read(buffer, offset, count, false).GetAwaiter().GetResult(); /// <summary> /// Reads <i>count</i> bytes from the large object. The only case when fewer bytes are read is when end of stream is reached. /// </summary> /// <param name="buffer">The buffer where read data should be stored.</param> /// <param name="offset">The offset in the buffer where the first byte should be read.</param> /// <param name="count">The maximum number of bytes that should be read.</param> /// <param name="cancellationToken"> /// An optional token to cancel the asynchronous operation. The default value is <see cref="CancellationToken.None"/>. /// </param> /// <returns>How many bytes actually read, or 0 if end of file was already reached.</returns> public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) { using (NoSynchronizationContextScope.Enter()) return Read(buffer, offset, count, true, cancellationToken); } async Task<int> Read(byte[] buffer, int offset, int count, bool async, CancellationToken cancellationToken = default) { if (buffer == null) throw new ArgumentNullException(nameof(buffer)); if (offset < 0) throw new ArgumentOutOfRangeException(nameof(offset)); if (count < 0) throw new ArgumentOutOfRangeException(nameof(count)); if (buffer.Length - offset < count) throw new ArgumentException("Invalid offset or count for this buffer"); CheckDisposed(); var chunkCount = Math.Min(count, _manager.MaxTransferBlockSize); var read = 0; while (read < count) { var bytesRead = await _manager.ExecuteFunctionGetBytes( "loread", buffer, offset + read, count - read, async, cancellationToken, _fd, chunkCount); _pos += bytesRead; read += bytesRead; if (bytesRead < chunkCount) { return read; } } return read; } /// <summary> /// Writes <i>count</i> bytes to the large object. /// </summary> /// <param name="buffer">The buffer to write data from.</param> /// <param name="offset">The offset in the buffer at which to begin copying bytes.</param> /// <param name="count">The number of bytes to write.</param> public override void Write(byte[] buffer, int offset, int count) => Write(buffer, offset, count, false).GetAwaiter().GetResult(); /// <summary> /// Writes <i>count</i> bytes to the large object. /// </summary> /// <param name="buffer">The buffer to write data from.</param> /// <param name="offset">The offset in the buffer at which to begin copying bytes.</param> /// <param name="count">The number of bytes to write.</param> /// <param name="cancellationToken"> /// An optional token to cancel the asynchronous operation. The default value is <see cref="CancellationToken.None"/>. /// </param> public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) { using (NoSynchronizationContextScope.Enter()) return Write(buffer, offset, count, true, cancellationToken); } async Task Write(byte[] buffer, int offset, int count, bool async, CancellationToken cancellationToken = default) { if (buffer == null) throw new ArgumentNullException(nameof(buffer)); if (offset < 0) throw new ArgumentOutOfRangeException(nameof(offset)); if (count < 0) throw new ArgumentOutOfRangeException(nameof(count)); if (buffer.Length - offset < count) throw new ArgumentException("Invalid offset or count for this buffer"); CheckDisposed(); if (!_writeable) throw new NotSupportedException("Write cannot be called on a stream opened with no write permissions"); var totalWritten = 0; while (totalWritten < count) { var chunkSize = Math.Min(count - totalWritten, _manager.MaxTransferBlockSize); var bytesWritten = await _manager.ExecuteFunction<int>("lowrite", async, cancellationToken, _fd, new ArraySegment<byte>(buffer, offset + totalWritten, chunkSize)); totalWritten += bytesWritten; if (bytesWritten != chunkSize) throw new InvalidOperationException($"Internal Npgsql bug, please report"); _pos += bytesWritten; } } /// <summary> /// CanTimeout always returns false. /// </summary> public override bool CanTimeout => false; /// <summary> /// CanRead always returns true, unless the stream has been closed. /// </summary> public override bool CanRead => !_disposed; /// <summary> /// CanWrite returns true if the stream was opened with write permissions, and the stream has not been closed. /// </summary> public override bool CanWrite => _writeable && !_disposed; /// <summary> /// CanSeek always returns true, unless the stream has been closed. /// </summary> public override bool CanSeek => !_disposed; /// <summary> /// Returns the current position in the stream. Getting the current position does not need a round-trip to the server, however setting the current position does. /// </summary> public override long Position { get { CheckDisposed(); return _pos; } set => Seek(value, SeekOrigin.Begin); } /// <summary> /// Gets the length of the large object. This internally seeks to the end of the stream to retrieve the length, and then back again. /// </summary> public override long Length => GetLength(false).GetAwaiter().GetResult(); /// <summary> /// Gets the length of the large object. This internally seeks to the end of the stream to retrieve the length, and then back again. /// </summary> /// <param name="cancellationToken"> /// An optional token to cancel the asynchronous operation. The default value is <see cref="CancellationToken.None"/>. /// </param> public Task<long> GetLengthAsync(CancellationToken cancellationToken = default) { using (NoSynchronizationContextScope.Enter()) return GetLength(true); } async Task<long> GetLength(bool async) { CheckDisposed(); var old = _pos; var retval = await Seek(0, SeekOrigin.End, async); if (retval != old) await Seek(old, SeekOrigin.Begin, async); return retval; } /// <summary> /// Seeks in the stream to the specified position. This requires a round-trip to the backend. /// </summary> /// <param name="offset">A byte offset relative to the <i>origin</i> parameter.</param> /// <param name="origin">A value of type SeekOrigin indicating the reference point used to obtain the new position.</param> /// <returns></returns> public override long Seek(long offset, SeekOrigin origin) => Seek(offset, origin, false).GetAwaiter().GetResult(); /// <summary> /// Seeks in the stream to the specified position. This requires a round-trip to the backend. /// </summary> /// <param name="offset">A byte offset relative to the <i>origin</i> parameter.</param> /// <param name="origin">A value of type SeekOrigin indicating the reference point used to obtain the new position.</param> /// <param name="cancellationToken"> /// An optional token to cancel the asynchronous operation. The default value is <see cref="CancellationToken.None"/>. /// </param> public Task<long> SeekAsync(long offset, SeekOrigin origin, CancellationToken cancellationToken = default) { using (NoSynchronizationContextScope.Enter()) return Seek(offset, origin, true, cancellationToken); } async Task<long> Seek(long offset, SeekOrigin origin, bool async, CancellationToken cancellationToken = default) { if (origin < SeekOrigin.Begin || origin > SeekOrigin.End) throw new ArgumentException("Invalid origin"); if (!Has64BitSupport && offset != (int)offset) throw new ArgumentOutOfRangeException(nameof(offset), "offset must fit in 32 bits for PostgreSQL versions older than 9.3"); CheckDisposed(); return _manager.Has64BitSupport ? _pos = await _manager.ExecuteFunction<long>("lo_lseek64", async, cancellationToken, _fd, offset, (int)origin) : _pos = await _manager.ExecuteFunction<int>("lo_lseek", async, cancellationToken, _fd, (int)offset, (int)origin); } /// <summary> /// Does nothing. /// </summary> public override void Flush() {} /// <summary> /// Truncates or enlarges the large object to the given size. If enlarging, the large object is extended with null bytes. /// For PostgreSQL versions earlier than 9.3, the value must fit in an Int32. /// </summary> /// <param name="value">Number of bytes to either truncate or enlarge the large object.</param> public override void SetLength(long value) => SetLength(value, false).GetAwaiter().GetResult(); /// <summary> /// Truncates or enlarges the large object to the given size. If enlarging, the large object is extended with null bytes. /// For PostgreSQL versions earlier than 9.3, the value must fit in an Int32. /// </summary> /// <param name="value">Number of bytes to either truncate or enlarge the large object.</param> /// <param name="cancellationToken"> /// An optional token to cancel the asynchronous operation. The default value is <see cref="CancellationToken.None"/>. /// </param> public Task SetLength(long value, CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); using (NoSynchronizationContextScope.Enter()) return SetLength(value, true, cancellationToken); } async Task SetLength(long value, bool async, CancellationToken cancellationToken = default) { if (value < 0) throw new ArgumentOutOfRangeException(nameof(value)); if (!Has64BitSupport && value != (int)value) throw new ArgumentOutOfRangeException(nameof(value), "offset must fit in 32 bits for PostgreSQL versions older than 9.3"); CheckDisposed(); if (!_writeable) throw new NotSupportedException("SetLength cannot be called on a stream opened with no write permissions"); if (_manager.Has64BitSupport) await _manager.ExecuteFunction<int>("lo_truncate64", async, cancellationToken, _fd, value); else await _manager.ExecuteFunction<int>("lo_truncate", async, cancellationToken, _fd, (int)value); } /// <summary> /// Releases resources at the backend allocated for this stream. /// </summary> public override void Close() { if (!_disposed) { _manager.ExecuteFunction<int>("lo_close", false, CancellationToken.None, _fd).GetAwaiter().GetResult(); _disposed = true; } } /// <summary> /// Releases resources at the backend allocated for this stream, iff disposing is true. /// </summary> /// <param name="disposing">Whether to release resources allocated at the backend.</param> protected override void Dispose(bool disposing) { if (disposing) { Close(); } } } } |
| You are visitor 01402520 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |