using System;
using Npgsql.Util;
using System.Data;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace Npgsql;
///
/// Large object manager. This class can be used to store very large files in a PostgreSQL database.
///
[Obsolete("NpgsqlLargeObjectManager allows manipulating PostgreSQL large objects via publicly available PostgreSQL functions (lo_read, lo_write); call these yourself directly.")]
public class NpgsqlLargeObjectManager
{
const int InvWrite = 0x00020000;
const int InvRead = 0x00040000;
internal NpgsqlConnection Connection { get; }
///
/// The largest chunk size (in bytes) read and write operations will read/write each roundtrip to the network. Default 4 MB.
///
public int MaxTransferBlockSize { get; set; }
///
/// Creates an NpgsqlLargeObjectManager for this connection. The connection must be opened to perform remote operations.
///
///
public NpgsqlLargeObjectManager(NpgsqlConnection connection)
{
Connection = connection;
MaxTransferBlockSize = 4 * 1024 * 1024; // 4MB
}
///
/// Execute a function
///
internal async Task ExecuteFunction(bool async, string function, CancellationToken cancellationToken, params object[] arguments)
{
using var command = Connection.CreateCommand();
var stringBuilder = new StringBuilder("SELECT * FROM ").Append(function).Append('(');
for (var i = 0; i < arguments.Length; i++)
{
if (i > 0)
stringBuilder.Append(", ");
stringBuilder.Append('$').Append(i + 1);
command.Parameters.Add(new NpgsqlParameter { Value = arguments[i] });
}
stringBuilder.Append(')');
command.CommandText = stringBuilder.ToString();
return (T)(async ? await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) : command.ExecuteScalar())!;
}
///
/// Execute a function that returns a byte array
///
///
internal async Task ExecuteFunctionGetBytes(
bool async, string function, byte[] buffer, int offset, int len, CancellationToken cancellationToken, params object[] arguments)
{
using var command = Connection.CreateCommand();
var stringBuilder = new StringBuilder("SELECT * FROM ").Append(function).Append('(');
for (var i = 0; i < arguments.Length; i++)
{
if (i > 0)
stringBuilder.Append(", ");
stringBuilder.Append('$').Append(i + 1);
command.Parameters.Add(new NpgsqlParameter { Value = arguments[i] });
}
stringBuilder.Append(')');
command.CommandText = stringBuilder.ToString();
var reader = async
? await command.ExecuteReaderAsync(CommandBehavior.SequentialAccess, cancellationToken).ConfigureAwait(false)
: command.ExecuteReader(CommandBehavior.SequentialAccess);
try
{
if (async)
await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
else
reader.Read();
return (int)reader.GetBytes(0, 0, buffer, offset, len);
}
finally
{
if (async)
await reader.DisposeAsync().ConfigureAwait(false);
else
reader.Dispose();
}
}
///
/// Create an empty large object in the database. If an oid is specified but is already in use, an PostgresException will be thrown.
///
/// A preferred oid, or specify 0 if one should be automatically assigned
/// The oid for the large object created
/// If an oid is already in use
public uint Create(uint preferredOid = 0) => Create(preferredOid, false).GetAwaiter().GetResult();
// Review unused parameters
///
/// Create an empty large object in the database. If an oid is specified but is already in use, an PostgresException will be thrown.
///
/// A preferred oid, or specify 0 if one should be automatically assigned
///
/// An optional token to cancel the asynchronous operation. The default value is .
///
/// The oid for the large object created
/// If an oid is already in use
public Task CreateAsync(uint preferredOid, CancellationToken cancellationToken = default)
=> Create(preferredOid, true, cancellationToken);
Task Create(uint preferredOid, bool async, CancellationToken cancellationToken = default)
=> ExecuteFunction(async, "lo_create", cancellationToken, (int)preferredOid);
///
/// Opens a large object on the backend, returning a stream controlling this remote object.
/// A transaction snapshot is taken by the backend when the object is opened with only read permissions.
/// When reading from this object, the contents reflects the time when the snapshot was taken.
/// Note that this method, as well as operations on the stream must be wrapped inside a transaction.
///
/// Oid of the object
/// An NpgsqlLargeObjectStream
public NpgsqlLargeObjectStream OpenRead(uint oid)
=> OpenRead(async: false, oid).GetAwaiter().GetResult();
///
/// Opens a large object on the backend, returning a stream controlling this remote object.
/// A transaction snapshot is taken by the backend when the object is opened with only read permissions.
/// When reading from this object, the contents reflects the time when the snapshot was taken.
/// Note that this method, as well as operations on the stream must be wrapped inside a transaction.
///
/// Oid of the object
///
/// An optional token to cancel the asynchronous operation. The default value is .
///
/// An NpgsqlLargeObjectStream
public Task OpenReadAsync(uint oid, CancellationToken cancellationToken = default)
=> OpenRead(async: true, oid, cancellationToken);
async Task OpenRead(bool async, uint oid, CancellationToken cancellationToken = default)
{
var fd = await ExecuteFunction(async, "lo_open", cancellationToken, (int)oid, InvRead).ConfigureAwait(false);
return new NpgsqlLargeObjectStream(this, fd, false);
}
///
/// Opens a large object on the backend, returning a stream controlling this remote object.
/// Note that this method, as well as operations on the stream must be wrapped inside a transaction.
///
/// Oid of the object
/// An NpgsqlLargeObjectStream
public NpgsqlLargeObjectStream OpenReadWrite(uint oid)
=> OpenReadWrite(async: false, oid).GetAwaiter().GetResult();
///
/// Opens a large object on the backend, returning a stream controlling this remote object.
/// Note that this method, as well as operations on the stream must be wrapped inside a transaction.
///
/// Oid of the object
///
/// An optional token to cancel the asynchronous operation. The default value is .
///
/// An NpgsqlLargeObjectStream
public Task OpenReadWriteAsync(uint oid, CancellationToken cancellationToken = default)
=> OpenReadWrite(async: true, oid, cancellationToken);
async Task OpenReadWrite(bool async, uint oid, CancellationToken cancellationToken = default)
{
var fd = await ExecuteFunction(async, "lo_open", cancellationToken, (int)oid, InvRead | InvWrite).ConfigureAwait(false);
return new NpgsqlLargeObjectStream(this, fd, true);
}
///
/// Deletes a large object on the backend.
///
/// Oid of the object to delete
public void Unlink(uint oid)
=> ExecuteFunction