Skip to content

Commit 93124b7

Browse files
vonzshikvonzshik
committed
Passing the cancellation token to the write buffer
1 parent 627f99a commit 93124b7

42 files changed

Lines changed: 516 additions & 474 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎src/Npgsql.GeoJSON/GeoJSONHandler.cs‎

Lines changed: 36 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -454,20 +454,20 @@ int INpgsqlTypeHandler<IGeoJSONObject>.ValidateAndGetLength(IGeoJSONObject value
454454
int INpgsqlTypeHandler<IGeometryObject>.ValidateAndGetLength(IGeometryObject value, ref NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter)
455455
=> ValidateAndGetLength((GeoJSONObject)value, ref lengthCache, parameter);
456456

457-
public override Task Write(GeoJSONObject value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
457+
public override Task Write(GeoJSONObject value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
458458
=> value.Type switch
459459
{
460-
GeoJSONObjectType.Point => Write((Point)value, buf, lengthCache, parameter, async),
461-
GeoJSONObjectType.LineString => Write((LineString)value, buf, lengthCache, parameter, async),
462-
GeoJSONObjectType.Polygon => Write((Polygon)value, buf, lengthCache, parameter, async),
463-
GeoJSONObjectType.MultiPoint => Write((MultiPoint)value, buf, lengthCache, parameter, async),
464-
GeoJSONObjectType.MultiLineString => Write((MultiLineString)value, buf, lengthCache, parameter, async),
465-
GeoJSONObjectType.MultiPolygon => Write((MultiPolygon)value, buf, lengthCache, parameter, async),
466-
GeoJSONObjectType.GeometryCollection => Write((GeometryCollection)value, buf, lengthCache, parameter, async),
460+
GeoJSONObjectType.Point => Write((Point)value, buf, lengthCache, parameter, async, cancellationToken),
461+
GeoJSONObjectType.LineString => Write((LineString)value, buf, lengthCache, parameter, async, cancellationToken),
462+
GeoJSONObjectType.Polygon => Write((Polygon)value, buf, lengthCache, parameter, async, cancellationToken),
463+
GeoJSONObjectType.MultiPoint => Write((MultiPoint)value, buf, lengthCache, parameter, async, cancellationToken),
464+
GeoJSONObjectType.MultiLineString => Write((MultiLineString)value, buf, lengthCache, parameter, async, cancellationToken),
465+
GeoJSONObjectType.MultiPolygon => Write((MultiPolygon)value, buf, lengthCache, parameter, async, cancellationToken),
466+
GeoJSONObjectType.GeometryCollection => Write((GeometryCollection)value, buf, lengthCache, parameter, async, cancellationToken),
467467
_ => throw UnknownPostGisType()
468468
};
469469

470-
public async Task Write(Point value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
470+
public async Task Write(Point value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
471471
{
472472
var type = EwkbGeometryType.Point;
473473
var size = SizeOfHeader;
@@ -479,18 +479,18 @@ public async Task Write(Point value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? l
479479
}
480480

481481
if (buf.WriteSpaceLeft < size)
482-
await buf.Flush(async);
482+
await buf.Flush(async, cancellationToken);
483483

484484
buf.WriteByte(0); // Most significant byte first
485485
buf.WriteInt32((int)type);
486486

487487
if (srid != 0)
488488
buf.WriteInt32(srid);
489489

490-
await WritePosition(value.Coordinates, buf, async);
490+
await WritePosition(value.Coordinates, buf, async, cancellationToken);
491491
}
492492

493-
public async Task Write(LineString value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
493+
public async Task Write(LineString value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
494494
{
495495
var type = EwkbGeometryType.LineString;
496496
var size = SizeOfHeader;
@@ -502,7 +502,7 @@ public async Task Write(LineString value, NpgsqlWriteBuffer buf, NpgsqlLengthCac
502502
}
503503

504504
if (buf.WriteSpaceLeft < size)
505-
await buf.Flush(async);
505+
await buf.Flush(async, cancellationToken);
506506

507507
var coordinates = value.Coordinates;
508508

@@ -514,10 +514,10 @@ public async Task Write(LineString value, NpgsqlWriteBuffer buf, NpgsqlLengthCac
514514
buf.WriteInt32(srid);
515515

516516
for (var i = 0; i < coordinates.Count; ++i)
517-
await WritePosition(coordinates[i], buf, async);
517+
await WritePosition(coordinates[i], buf, async, cancellationToken);
518518
}
519519

520-
public async Task Write(Polygon value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
520+
public async Task Write(Polygon value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
521521
{
522522
var type = EwkbGeometryType.Polygon;
523523
var size = SizeOfHeader;
@@ -529,7 +529,7 @@ public async Task Write(Polygon value, NpgsqlWriteBuffer buf, NpgsqlLengthCache?
529529
}
530530

531531
if (buf.WriteSpaceLeft < size)
532-
await buf.Flush(async);
532+
await buf.Flush(async, cancellationToken);
533533

534534
var lines = value.Coordinates;
535535

@@ -543,15 +543,15 @@ public async Task Write(Polygon value, NpgsqlWriteBuffer buf, NpgsqlLengthCache?
543543
for (var i = 0; i < lines.Count; ++i)
544544
{
545545
if (buf.WriteSpaceLeft < 4)
546-
await buf.Flush(async);
546+
await buf.Flush(async, cancellationToken);
547547
var coordinates = lines[i].Coordinates;
548548
buf.WriteInt32(coordinates.Count);
549549
for (var j = 0; j < coordinates.Count; ++j)
550-
await WritePosition(coordinates[j], buf, async);
550+
await WritePosition(coordinates[j], buf, async, cancellationToken);
551551
}
552552
}
553553

554-
public async Task Write(MultiPoint value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
554+
public async Task Write(MultiPoint value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
555555
{
556556
var type = EwkbGeometryType.MultiPoint;
557557
var size = SizeOfHeader;
@@ -563,7 +563,7 @@ public async Task Write(MultiPoint value, NpgsqlWriteBuffer buf, NpgsqlLengthCac
563563
}
564564

565565
if (buf.WriteSpaceLeft < size)
566-
await buf.Flush(async);
566+
await buf.Flush(async, cancellationToken);
567567

568568
var coordinates = value.Coordinates;
569569

@@ -575,10 +575,10 @@ public async Task Write(MultiPoint value, NpgsqlWriteBuffer buf, NpgsqlLengthCac
575575
buf.WriteInt32(srid);
576576

577577
for (var i = 0; i < coordinates.Count; ++i)
578-
await Write(coordinates[i], buf, lengthCache, parameter, async);
578+
await Write(coordinates[i], buf, lengthCache, parameter, async, cancellationToken);
579579
}
580580

581-
public async Task Write(MultiLineString value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
581+
public async Task Write(MultiLineString value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
582582
{
583583
var type = EwkbGeometryType.MultiLineString;
584584
var size = SizeOfHeader;
@@ -590,7 +590,7 @@ public async Task Write(MultiLineString value, NpgsqlWriteBuffer buf, NpgsqlLeng
590590
}
591591

592592
if (buf.WriteSpaceLeft < size)
593-
await buf.Flush(async);
593+
await buf.Flush(async, cancellationToken);
594594

595595
var coordinates = value.Coordinates;
596596

@@ -602,10 +602,10 @@ public async Task Write(MultiLineString value, NpgsqlWriteBuffer buf, NpgsqlLeng
602602
buf.WriteInt32(srid);
603603

604604
for (var i = 0; i < coordinates.Count; ++i)
605-
await Write(coordinates[i], buf, lengthCache, parameter, async);
605+
await Write(coordinates[i], buf, lengthCache, parameter, async, cancellationToken);
606606
}
607607

608-
public async Task Write(MultiPolygon value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
608+
public async Task Write(MultiPolygon value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
609609
{
610610
var type = EwkbGeometryType.MultiPolygon;
611611
var size = SizeOfHeader;
@@ -617,7 +617,7 @@ public async Task Write(MultiPolygon value, NpgsqlWriteBuffer buf, NpgsqlLengthC
617617
}
618618

619619
if (buf.WriteSpaceLeft < size)
620-
await buf.Flush(async);
620+
await buf.Flush(async, cancellationToken);
621621

622622
var coordinates = value.Coordinates;
623623

@@ -628,10 +628,10 @@ public async Task Write(MultiPolygon value, NpgsqlWriteBuffer buf, NpgsqlLengthC
628628
if (srid != 0)
629629
buf.WriteInt32(srid);
630630
for (var i = 0; i < coordinates.Count; ++i)
631-
await Write(coordinates[i], buf, lengthCache, parameter, async);
631+
await Write(coordinates[i], buf, lengthCache, parameter, async, cancellationToken);
632632
}
633633

634-
public async Task Write(GeometryCollection value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
634+
public async Task Write(GeometryCollection value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
635635
{
636636
var type = EwkbGeometryType.GeometryCollection;
637637
var size = SizeOfHeader;
@@ -643,7 +643,7 @@ public async Task Write(GeometryCollection value, NpgsqlWriteBuffer buf, NpgsqlL
643643
}
644644

645645
if (buf.WriteSpaceLeft < size)
646-
await buf.Flush(async);
646+
await buf.Flush(async, cancellationToken);
647647

648648
var geometries = value.Geometries;
649649

@@ -655,20 +655,20 @@ public async Task Write(GeometryCollection value, NpgsqlWriteBuffer buf, NpgsqlL
655655
buf.WriteInt32(srid);
656656

657657
for (var i = 0; i < geometries.Count; ++i)
658-
await Write((GeoJSONObject)geometries[i], buf, lengthCache, parameter, async);
658+
await Write((GeoJSONObject) geometries[i], buf, lengthCache, parameter, async, cancellationToken);
659659
}
660660

661-
Task INpgsqlTypeHandler<IGeoJSONObject>.Write(IGeoJSONObject value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
662-
=> Write((GeoJSONObject)value, buf, lengthCache, parameter, async);
661+
Task INpgsqlTypeHandler<IGeoJSONObject>.Write(IGeoJSONObject value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken)
662+
=> Write((GeoJSONObject)value, buf, lengthCache, parameter, async, cancellationToken);
663663

664-
Task INpgsqlTypeHandler<IGeometryObject>.Write(IGeometryObject value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
665-
=> Write((GeoJSONObject)value, buf, lengthCache, parameter, async);
664+
Task INpgsqlTypeHandler<IGeometryObject>.Write(IGeometryObject value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken)
665+
=> Write((GeoJSONObject)value, buf, lengthCache, parameter, async, cancellationToken);
666666

667-
static async Task WritePosition(IPosition coordinate, NpgsqlWriteBuffer buf, bool async)
667+
static async Task WritePosition(IPosition coordinate, NpgsqlWriteBuffer buf, bool async, CancellationToken cancellationToken = default)
668668
{
669669
var altitude = coordinate.Altitude;
670670
if (buf.WriteSpaceLeft < SizeOfPoint(altitude.HasValue))
671-
await buf.Flush(async);
671+
await buf.Flush(async, cancellationToken);
672672
buf.WriteDouble(coordinate.Longitude);
673673
buf.WriteDouble(coordinate.Latitude);
674674
if (altitude.HasValue)

‎src/Npgsql.Json.NET/JsonHandler.cs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -59,22 +59,22 @@ protected override int ValidateAndGetLength<T2>(T2 value, ref NpgsqlLengthCache?
5959
return base.ValidateAndGetLength(serialized, ref lengthCache, parameter);
6060
}
6161

62-
protected override Task WriteWithLength<T2>(T2 value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
62+
protected override Task WriteWithLength<T2>(T2 value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
6363
{
6464
if (typeof(T2) == typeof(string) ||
6565
typeof(T2) == typeof(char[]) ||
6666
typeof(T2) == typeof(ArraySegment<char>) ||
6767
typeof(T2) == typeof(char) ||
6868
typeof(T2) == typeof(byte[]))
6969
{
70-
return base.WriteWithLength(value, buf, lengthCache, parameter, async);
70+
return base.WriteWithLength(value, buf, lengthCache, parameter, async, cancellationToken);
7171
}
7272

7373
// User POCO, read serialized representation from the validation phase
7474
var serialized = parameter?.ConvertedValue != null
7575
? (string)parameter.ConvertedValue
7676
: JsonConvert.SerializeObject(value, _settings);
77-
return base.WriteWithLength(serialized, buf, lengthCache, parameter, async);
77+
return base.WriteWithLength(serialized, buf, lengthCache, parameter, async, cancellationToken);
7878
}
7979

8080
protected override int ValidateObjectAndGetLength(object value, ref NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter)
@@ -92,7 +92,7 @@ value is char ||
9292
return ValidateAndGetLength(value, ref lengthCache, parameter);
9393
}
9494

95-
protected override Task WriteObjectWithLength(object value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
95+
protected override Task WriteObjectWithLength(object value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
9696
{
9797
if (value is DBNull ||
9898
value is string ||
@@ -101,10 +101,10 @@ value is ArraySegment<char> ||
101101
value is char ||
102102
value is byte[])
103103
{
104-
return base.WriteObjectWithLength(value, buf, lengthCache, parameter, async);
104+
return base.WriteObjectWithLength(value, buf, lengthCache, parameter, async, cancellationToken);
105105
}
106106

107-
return WriteWithLength(value, buf, lengthCache, parameter, async);
107+
return WriteWithLength(value, buf, lengthCache, parameter, async, cancellationToken);
108108
}
109109
}
110110
}

‎src/Npgsql.Json.NET/JsonbHandler.cs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -59,22 +59,22 @@ protected override int ValidateAndGetLength<T2>(T2 value, ref NpgsqlLengthCache?
5959
return base.ValidateAndGetLength(serialized, ref lengthCache, parameter);
6060
}
6161

62-
protected override Task WriteWithLength<T2>(T2 value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
62+
protected override Task WriteWithLength<T2>(T2 value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
6363
{
6464
if (typeof(T2) == typeof(string) ||
6565
typeof(T2) == typeof(char[]) ||
6666
typeof(T2) == typeof(ArraySegment<char>) ||
6767
typeof(T2) == typeof(char) ||
6868
typeof(T2) == typeof(byte[]))
6969
{
70-
return base.WriteWithLength(value, buf, lengthCache, parameter, async);
70+
return base.WriteWithLength(value, buf, lengthCache, parameter, async, cancellationToken);
7171
}
7272

7373
// User POCO, read serialized representation from the validation phase
7474
var serialized = parameter?.ConvertedValue != null
7575
? (string)parameter.ConvertedValue
7676
: JsonConvert.SerializeObject(value, _settings);
77-
return base.WriteWithLength(serialized, buf, lengthCache, parameter, async);
77+
return base.WriteWithLength(serialized, buf, lengthCache, parameter, async, cancellationToken);
7878
}
7979

8080
protected override int ValidateObjectAndGetLength(object value, ref NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter)
@@ -92,7 +92,7 @@ value is char ||
9292
return ValidateAndGetLength(value, ref lengthCache, parameter);
9393
}
9494

95-
protected override Task WriteObjectWithLength(object value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async)
95+
protected override Task WriteObjectWithLength(object value, NpgsqlWriteBuffer buf, NpgsqlLengthCache? lengthCache, NpgsqlParameter? parameter, bool async, CancellationToken cancellationToken = default)
9696
{
9797
if (value is DBNull ||
9898
value is string ||
@@ -101,10 +101,10 @@ value is ArraySegment<char> ||
101101
value is char ||
102102
value is byte[])
103103
{
104-
return base.WriteObjectWithLength(value, buf, lengthCache, parameter, async);
104+
return base.WriteObjectWithLength(value, buf, lengthCache, parameter, async, cancellationToken);
105105
}
106106

107-
return WriteWithLength(value, buf, lengthCache, parameter, async);
107+
return WriteWithLength(value, buf, lengthCache, parameter, async, cancellationToken);
108108
}
109109
}
110110
}

0 commit comments

Comments
 (0)