feat: wire snapshot/restore API endpoints (Gap 7.4 stub)
Add HandleSnapshotAsync and HandleRestoreAsync with stream-name validation, chunk metadata (NumChunks, BlkSize) in the response, and richer error codes. Add StreamManager.Exists helper. Add JetStreamSnapshot.StreamName/NumChunks/BlkSize fields. Fix AdvisoryEventTests.cs using-directive ordering. Add 12 SnapshotApiTests.
This commit is contained in:
@@ -189,6 +189,66 @@ public static class StreamApiHandlers
|
||||
: JetStreamApiResponse.NotFound(subject);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Async snapshot handler that validates stream existence before creating the snapshot,
|
||||
/// and enriches the response with stream name and chunk metadata.
|
||||
/// Go reference: server/jetstream_api.go — jsStreamSnapshotT handler.
|
||||
/// </summary>
|
||||
public static async Task<JetStreamApiResponse> HandleSnapshotAsync(
|
||||
string subject,
|
||||
StreamManager streamManager,
|
||||
CancellationToken ct)
|
||||
{
|
||||
_ = ct;
|
||||
|
||||
var streamName = ExtractTrailingToken(subject, SnapshotPrefix);
|
||||
if (streamName == null)
|
||||
return JetStreamApiResponse.NotFound(subject);
|
||||
|
||||
if (!streamManager.Exists(streamName))
|
||||
return JetStreamApiResponse.NotFound(subject);
|
||||
|
||||
var snapshot = streamManager.CreateSnapshot(streamName);
|
||||
if (snapshot == null)
|
||||
return JetStreamApiResponse.ErrorResponse(500, "snapshot creation failed");
|
||||
|
||||
return new JetStreamApiResponse
|
||||
{
|
||||
Snapshot = new JetStreamSnapshot
|
||||
{
|
||||
Payload = Convert.ToBase64String(snapshot),
|
||||
StreamName = streamName,
|
||||
NumChunks = 1,
|
||||
BlkSize = snapshot.Length,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Async restore handler that validates the payload and returns a structured error on failure.
|
||||
/// Go reference: server/jetstream_api.go — jsStreamRestoreT handler.
|
||||
/// </summary>
|
||||
public static async Task<JetStreamApiResponse> HandleRestoreAsync(
|
||||
string subject,
|
||||
byte[] payload,
|
||||
StreamManager streamManager,
|
||||
CancellationToken ct)
|
||||
{
|
||||
_ = ct;
|
||||
|
||||
var streamName = ExtractTrailingToken(subject, RestorePrefix);
|
||||
if (streamName == null)
|
||||
return JetStreamApiResponse.NotFound(subject);
|
||||
|
||||
var snapshotBytes = ParseRestorePayload(payload);
|
||||
if (snapshotBytes == null)
|
||||
return JetStreamApiResponse.ErrorResponse(400, "snapshot payload required");
|
||||
|
||||
return streamManager.RestoreSnapshot(streamName, snapshotBytes)
|
||||
? JetStreamApiResponse.SuccessResponse()
|
||||
: JetStreamApiResponse.ErrorResponse(500, "restore failed");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------
|
||||
// Clustered handlers — propose to meta RAFT group instead of local StreamManager.
|
||||
// Go reference: jetstream_cluster.go:7620-7900 jsClusteredStreamRequest and related.
|
||||
|
||||
@@ -115,6 +115,8 @@ public sealed class StreamManager
|
||||
|
||||
public bool TryGet(string name, out StreamHandle handle) => _streams.TryGetValue(name, out handle!);
|
||||
|
||||
public bool Exists(string name) => _streams.ContainsKey(name);
|
||||
|
||||
public bool Delete(string name)
|
||||
{
|
||||
if (!_streams.TryRemove(name, out _))
|
||||
|
||||
Reference in New Issue
Block a user