Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Joseph Doherty
|
d2d2e5f68f | ||
|
Joseph Doherty
|
d692232191 | ||
|
Joseph Doherty
|
65943597d4 | ||
|
Joseph Doherty
|
27ed65114e | ||
|
Joseph Doherty
|
397d3c5c4f | ||
|
Joseph Doherty
|
dc9c0c950c | ||
 dohertj2
|
867bf18116 |
@@ -32,7 +32,7 @@ dotnet test src/MxGateway.Worker.Tests/MxGateway.Worker.Tests.csproj -p:Platform
|
||||
dotnet run --project src/MxGateway.Server/MxGateway.Server.csproj
|
||||
|
||||
# API-key admin CLI (same exe, "apikey" subcommand)
|
||||
dotnet run --project src/MxGateway.Server/MxGateway.Server.csproj -- apikey create --display-name "dev" --scopes session:open,session:close,invoke:read,invoke:write,invoke:secure,events:read,metadata:read,admin
|
||||
dotnet run --project src/MxGateway.Server/MxGateway.Server.csproj -- apikey create --display-name "dev" --scopes session,invoke,event,metadata,admin
|
||||
```
|
||||
|
||||
Single test by name (xUnit `--filter`):
|
||||
@@ -114,9 +114,9 @@ External analysis sources referenced by design docs:
|
||||
|
||||
## Authentication
|
||||
|
||||
Gateway gRPC clients authenticate with an API key in metadata: `authorization: Bearer mxgw_<key-id>_<secret>`. Keys are stored hashed (with a peppered SHA) in a gateway-owned SQLite DB (default `C:\ProgramData\MxGateway\gateway-auth.db`). Scopes (`session:open`, `session:close`, `invoke:read`, `invoke:write`, `invoke:secure`, `events:read`, `metadata:read`, `admin`) gate specific RPCs; missing → `Unauthenticated`, insufficient → `PermissionDenied`. The `apikey` subcommand on the server exe manages keys; see `src/MxGateway.Server/Security/Authentication/`.
|
||||
Gateway gRPC clients authenticate with an API key in metadata: `authorization: Bearer mxgw_<key-id>_<secret>`. Keys are stored hashed (with a peppered SHA) in a gateway-owned SQLite DB (default `C:\ProgramData\MxGateway\gateway-auth.db`). Scopes (`session`, `invoke`, `event`, `metadata`, `admin`) gate specific RPCs; missing → `Unauthenticated`, insufficient → `PermissionDenied`. The `apikey` subcommand on the server exe manages keys; see `src/MxGateway.Server/Security/Authentication/`.
|
||||
|
||||
Dashboard auth uses the same verifier but exchanges the API key for an HTTP-only secure cookie at `/dashboard/login`. `Dashboard:AllowAnonymousLocalhost` bypasses cookie auth on loopback when explicitly enabled.
|
||||
Dashboard auth is LDAP-backed (separate from the gRPC API-key model). `/login` binds against `MxGateway:Ldap` and maps the user's LDAP groups to `Admin` or `Viewer` via `MxGateway:Dashboard:GroupToRole`, then issues an HTTP-only secure `__Host-MxGatewayDashboard` cookie. SignalR hubs at `/hubs/{snapshot,alarms,events}` accept either the cookie or a 30-minute bearer minted at `/hubs/token`. `Dashboard:AllowAnonymousLocalhost` bypasses auth on loopback when enabled.
|
||||
|
||||
## Process / Platform Notes
|
||||
|
||||
|
||||
@@ -1,140 +0,0 @@
|
||||
# Code Review Process
|
||||
|
||||
This document describes how to perform a comprehensive, per-module code review of
|
||||
the `mxaccessgw` codebase and how to track findings to resolution.
|
||||
|
||||
A **module** is one buildable project under `src/` (e.g. `src/MxGateway.Worker`)
|
||||
or one language client under `clients/` (e.g. `clients/rust`). Each module has
|
||||
its own folder under `code-reviews/` containing a single `findings.md`.
|
||||
|
||||
## 1. Before you start
|
||||
|
||||
1. Pick the module to review. Its folder is `code-reviews/<Module>/`:
|
||||
- For a `src/` project, `<Module>` is the project name with the `MxGateway.`
|
||||
prefix stripped — `src/MxGateway.Server` is reviewed in `code-reviews/Server/`.
|
||||
- For a language client, `<Module>` is `Client.<Lang>` — `clients/rust` is
|
||||
reviewed in `code-reviews/Client.Rust/`.
|
||||
2. Identify the design context for the module:
|
||||
- `gateway.md` — top-level architecture, command/event surface, IPC envelope,
|
||||
STA thread model, fault handling.
|
||||
- The relevant component design docs under `docs/` (e.g.
|
||||
`docs/MxAccessWorkerInstanceDesign.md`, `docs/GatewayProcessDesign.md`,
|
||||
`docs/Sessions.md`, `docs/Authentication.md`, `docs/GalaxyRepository.md`).
|
||||
- `docs/DesignDecisions.md` for the v1 design choices.
|
||||
- The **Repository-Specific Conventions** and **Process / Platform Notes** in
|
||||
`CLAUDE.md`.
|
||||
3. Record the exact commit being reviewed: `git rev-parse --short HEAD`. Every
|
||||
review is a snapshot — a finding only means something relative to a known
|
||||
commit.
|
||||
4. Open `code-reviews/<Module>/findings.md` and fill in the header table
|
||||
(reviewer, date, commit SHA, status).
|
||||
|
||||
## 2. Review checklist
|
||||
|
||||
Work through **every** category below for the module. A comprehensive review
|
||||
means the checklist is completed even where it produces no findings — record
|
||||
"No issues found" for a category rather than leaving it ambiguous.
|
||||
|
||||
1. **Correctness & logic bugs** — off-by-one, null handling, incorrect
|
||||
conditionals, misuse of APIs, broken edge cases.
|
||||
2. **mxaccessgw conventions** — the rules in `CLAUDE.md` and the style guides
|
||||
under `docs/style-guides/`: the gateway never instantiates MXAccess COM
|
||||
directly; all MXAccess COM calls run on the worker's dedicated STA thread and
|
||||
the STA loop pumps Windows messages; IPC uses one bidirectional named pipe per
|
||||
worker carrying length-prefixed `WorkerEnvelope` protobuf frames; MXAccess
|
||||
parity is the contract (don't "fix" surprising MXAccess behaviour, never
|
||||
synthesize events); one worker and one event subscriber per session; the
|
||||
gateway terminates orphan workers on startup and does not reattach; C# style
|
||||
(file-scoped namespaces, `sealed` by default, `Async` suffix, MXAccess-aligned
|
||||
names); no Blazor UI component libraries; no logging of secrets or full tag
|
||||
values; generated code is never hand-edited.
|
||||
3. **Concurrency & thread safety** — shared mutable state, STA affinity, race
|
||||
conditions, correct use of `async`/`await`, locking, disposal races.
|
||||
4. **Error handling & resilience** — exception paths, worker crash / reconnect
|
||||
handling, fail-fast event backpressure, transient vs permanent error
|
||||
classification, graceful degradation, correct gRPC status codes.
|
||||
5. **Security** — authentication/authorization checks, API-key scope enforcement,
|
||||
input validation, SQL injection in the Galaxy Repository RPCs, secret
|
||||
handling, the dashboard anonymous-localhost bypass, logging of sensitive data.
|
||||
6. **Performance & resource management** — `IDisposable` disposal, pipe / stream
|
||||
/ COM lifetimes, buffering and back-pressure, unnecessary allocations on hot
|
||||
paths, N+1 queries.
|
||||
7. **Design-document adherence** — does the code match `gateway.md`, the relevant
|
||||
`docs/` component designs, `docs/DesignDecisions.md`, and `CLAUDE.md`? Flag
|
||||
both code that drifts from the design and design docs that are now stale.
|
||||
8. **Code organization & conventions** — namespace hierarchy, project layout, the
|
||||
Options pattern, separation of concerns, additive-only contract evolution.
|
||||
9. **Testing coverage** — are the module's behaviours covered by tests
|
||||
(`src/MxGateway.Tests`, `src/MxGateway.Worker.Tests`,
|
||||
`src/MxGateway.IntegrationTests`)? Note untested critical paths and missing
|
||||
edge-case tests.
|
||||
10. **Documentation & comments** — XML doc accuracy, misleading or stale comments,
|
||||
undocumented non-obvious behaviour.
|
||||
|
||||
## 3. Recording findings
|
||||
|
||||
Add one entry per finding to the `## Findings` section of the module's
|
||||
`findings.md`, using the entry format in
|
||||
[`_template/findings.md`](code-reviews/_template/findings.md).
|
||||
|
||||
- **Finding ID** — `<Module>-NNN`, numbered sequentially within the module and
|
||||
never reused (e.g. `Worker-001`). IDs are permanent even after resolution.
|
||||
- **Severity:**
|
||||
- **Critical** — data loss, security breach, crash/deadlock, or outage.
|
||||
- **High** — incorrect behaviour with significant impact; no safe workaround.
|
||||
- **Medium** — incorrect or risky behaviour with limited impact or a workaround.
|
||||
- **Low** — minor issues, style, maintainability, documentation.
|
||||
- **Category** — one of the 10 checklist categories above.
|
||||
- **Location** — `file:line` (clickable), or a list of locations.
|
||||
- **Description** — what is wrong and why it matters.
|
||||
- **Recommendation** — concrete suggested fix.
|
||||
|
||||
After recording findings, update the module header table (status, open-finding
|
||||
count) and regenerate the base README (step 5).
|
||||
|
||||
## 4. Marking an item resolved
|
||||
|
||||
Findings are **never deleted** — they are an audit trail. To close one, change
|
||||
its **Status** and complete the **Resolution** field:
|
||||
|
||||
- `Open` — newly recorded, not yet addressed.
|
||||
- `In Progress` — a fix is actively being worked on.
|
||||
- `Resolved` — fixed. The Resolution field must state the fixing commit SHA, the
|
||||
date, and a one-line description of the fix.
|
||||
- `Won't Fix` — intentionally not fixed. The Resolution field must justify why.
|
||||
- `Deferred` — valid but postponed. The Resolution field must say what it is
|
||||
waiting on (e.g. a tracked issue or a later milestone).
|
||||
|
||||
`Resolved`, `Won't Fix`, and `Deferred` findings are all considered **closed**.
|
||||
`Open` and `In Progress` are **pending** and appear in the base README's Pending
|
||||
Findings table.
|
||||
|
||||
## 5. Updating the base README
|
||||
|
||||
`code-reviews/README.md` holds the single cross-module view (the Module Status
|
||||
table and the Pending / Closed Findings tables). It is **generated** from the
|
||||
per-module `findings.md` files — do not edit it by hand.
|
||||
|
||||
After any review or status change, regenerate it:
|
||||
|
||||
```
|
||||
python code-reviews/regen-readme.py
|
||||
```
|
||||
|
||||
`regen-readme.py --check` exits non-zero if `README.md` is stale, if a module
|
||||
header's `Open findings` count disagrees with its finding statuses, or if a
|
||||
finding carries an unrecognised Status value. The PowerShell wrapper
|
||||
`scripts/check-code-reviews-readme.ps1` runs that check and is the intended hook
|
||||
for CI or a pre-commit step.
|
||||
|
||||
> The repo's installed `python` is the real interpreter; the bare `python3`
|
||||
> alias resolves to the Windows Store stub and fails. Use `python`.
|
||||
|
||||
The per-module `findings.md` files are the source of truth; `README.md` is the
|
||||
aggregated index and must always agree with them — which the script guarantees.
|
||||
|
||||
## 6. Re-reviewing a module
|
||||
|
||||
Re-reviews append to the same `findings.md`. Update the header to the new commit
|
||||
and date, continue the finding numbering from the last used ID, and leave prior
|
||||
findings (including closed ones) in place as history.
|
||||
@@ -16,9 +16,9 @@ Recommended layout:
|
||||
|
||||
```text
|
||||
clients/dotnet/
|
||||
MxGateway.Client.sln
|
||||
MxGateway.Client/
|
||||
MxGateway.Client.csproj
|
||||
ZB.MOM.WW.MxGateway.Client.slnx
|
||||
ZB.MOM.WW.MxGateway.Client/
|
||||
ZB.MOM.WW.MxGateway.Client.csproj
|
||||
GatewayClient.cs
|
||||
MxGatewaySession.cs
|
||||
MxGatewayClientOptions.cs
|
||||
@@ -26,14 +26,14 @@ clients/dotnet/
|
||||
Conversion/
|
||||
Errors/
|
||||
Generated/
|
||||
MxGateway.Client.Cli/
|
||||
MxGateway.Client.Cli.csproj
|
||||
ZB.MOM.WW.MxGateway.Client.Cli/
|
||||
ZB.MOM.WW.MxGateway.Client.Cli.csproj
|
||||
Program.cs
|
||||
Commands/
|
||||
MxGateway.Client.Tests/
|
||||
MxGateway.Client.Tests.csproj
|
||||
MxGateway.Client.IntegrationTests/
|
||||
MxGateway.Client.IntegrationTests.csproj
|
||||
ZB.MOM.WW.MxGateway.Client.Tests/
|
||||
ZB.MOM.WW.MxGateway.Client.Tests.csproj
|
||||
ZB.MOM.WW.MxGateway.Client.IntegrationTests/
|
||||
ZB.MOM.WW.MxGateway.Client.IntegrationTests.csproj
|
||||
```
|
||||
|
||||
Target framework:
|
||||
@@ -43,7 +43,7 @@ Target framework:
|
||||
```
|
||||
|
||||
The scaffold uses a project reference to
|
||||
`src/MxGateway.Contracts/MxGateway.Contracts.csproj` for generated protobuf and
|
||||
`src/ZB.MOM.WW.MxGateway.Contracts/ZB.MOM.WW.MxGateway.Contracts.csproj` for generated protobuf and
|
||||
gRPC types. `clients/dotnet/generated` remains reserved for client-local
|
||||
generator output if the .NET client later needs to decouple from the contracts
|
||||
project.
|
||||
@@ -166,7 +166,7 @@ reply.EnsureMxAccessSuccess();
|
||||
|
||||
## Test CLI
|
||||
|
||||
Project: `MxGateway.Client.Cli`.
|
||||
Project: `ZB.MOM.WW.MxGateway.Client.Cli`.
|
||||
|
||||
Command examples:
|
||||
|
||||
|
||||
@@ -1,76 +0,0 @@
|
||||
using Grpc.Core;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
|
||||
/// <summary>Tests for the shared gRPC-to-native exception mapping used by the transports.</summary>
|
||||
public sealed class RpcExceptionMapperTests
|
||||
{
|
||||
/// <summary>Verifies that an unauthenticated status maps to the authentication exception.</summary>
|
||||
[Fact]
|
||||
public void Map_UnauthenticatedStatus_ProducesAuthenticationException()
|
||||
{
|
||||
RpcException rpc = new(new Status(StatusCode.Unauthenticated, "no key"));
|
||||
|
||||
Exception mapped = RpcExceptionMapper.Map(rpc, CancellationToken.None);
|
||||
|
||||
MxGatewayAuthenticationException authentication =
|
||||
Assert.IsType<MxGatewayAuthenticationException>(mapped);
|
||||
Assert.Equal(StatusCode.Unauthenticated, authentication.StatusCode);
|
||||
}
|
||||
|
||||
/// <summary>Verifies that a permission-denied status maps to the authorization exception.</summary>
|
||||
[Fact]
|
||||
public void Map_PermissionDeniedStatus_ProducesAuthorizationException()
|
||||
{
|
||||
RpcException rpc = new(new Status(StatusCode.PermissionDenied, "missing scope"));
|
||||
|
||||
Exception mapped = RpcExceptionMapper.Map(rpc, CancellationToken.None);
|
||||
|
||||
MxGatewayAuthorizationException authorization =
|
||||
Assert.IsType<MxGatewayAuthorizationException>(mapped);
|
||||
Assert.Equal(StatusCode.PermissionDenied, authorization.StatusCode);
|
||||
}
|
||||
|
||||
/// <summary>Verifies that a cancelled status maps to OperationCanceledException.</summary>
|
||||
[Fact]
|
||||
public void Map_CancelledStatus_ProducesOperationCanceledException()
|
||||
{
|
||||
RpcException rpc = new(new Status(StatusCode.Cancelled, "cancelled"));
|
||||
|
||||
Exception mapped = RpcExceptionMapper.Map(rpc, CancellationToken.None);
|
||||
|
||||
Assert.IsType<OperationCanceledException>(mapped);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that non-auth statuses surface the originating gRPC status code on the
|
||||
/// mapped exception so callers can distinguish transient from permanent failures
|
||||
/// without reflecting into InnerException.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData(StatusCode.NotFound)]
|
||||
[InlineData(StatusCode.InvalidArgument)]
|
||||
[InlineData(StatusCode.ResourceExhausted)]
|
||||
[InlineData(StatusCode.FailedPrecondition)]
|
||||
[InlineData(StatusCode.Unavailable)]
|
||||
[InlineData(StatusCode.Internal)]
|
||||
public void Map_NonAuthStatus_CarriesStatusCodeOnMxGatewayException(StatusCode statusCode)
|
||||
{
|
||||
RpcException rpc = new(new Status(statusCode, "boom"));
|
||||
|
||||
Exception mapped = RpcExceptionMapper.Map(rpc, CancellationToken.None);
|
||||
|
||||
MxGatewayException gatewayException = Assert.IsType<MxGatewayException>(mapped);
|
||||
Assert.Equal(statusCode, gatewayException.StatusCode);
|
||||
Assert.Same(rpc, gatewayException.InnerException);
|
||||
}
|
||||
|
||||
/// <summary>Verifies that an MxGatewayException built without a gRPC status reports a null StatusCode.</summary>
|
||||
[Fact]
|
||||
public void StatusCode_IsNull_WhenNoGrpcStatusProvided()
|
||||
{
|
||||
MxGatewayException gatewayException = new("plain failure");
|
||||
|
||||
Assert.Null(gatewayException.StatusCode);
|
||||
}
|
||||
}
|
||||
@@ -1,76 +0,0 @@
|
||||
|
||||
Microsoft Visual Studio Solution File, Format Version 12.00
|
||||
# Visual Studio Version 17
|
||||
VisualStudioVersion = 17.0.31903.59
|
||||
MinimumVisualStudioVersion = 10.0.40219.1
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MxGateway.Client", "MxGateway.Client\MxGateway.Client.csproj", "{7CF9ED88-1F32-4040-BEB1-D0902E304C70}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MxGateway.Contracts", "..\..\src\MxGateway.Contracts\MxGateway.Contracts.csproj", "{9AB807A8-0469-40F7-A000-D240F36B6E5D}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MxGateway.Client.Cli", "MxGateway.Client.Cli\MxGateway.Client.Cli.csproj", "{EB061E77-2475-4322-9257-3F2456DD141C}"
|
||||
EndProject
|
||||
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "MxGateway.Client.Tests", "MxGateway.Client.Tests\MxGateway.Client.Tests.csproj", "{B77B5A8E-0C53-4419-9BCD-227C9753A074}"
|
||||
EndProject
|
||||
Global
|
||||
GlobalSection(SolutionConfigurationPlatforms) = preSolution
|
||||
Debug|Any CPU = Debug|Any CPU
|
||||
Debug|x64 = Debug|x64
|
||||
Debug|x86 = Debug|x86
|
||||
Release|Any CPU = Release|Any CPU
|
||||
Release|x64 = Release|x64
|
||||
Release|x86 = Release|x86
|
||||
EndGlobalSection
|
||||
GlobalSection(ProjectConfigurationPlatforms) = postSolution
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Debug|x64.ActiveCfg = Debug|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Debug|x64.Build.0 = Debug|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Debug|x86.ActiveCfg = Debug|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Debug|x86.Build.0 = Debug|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Release|x64.ActiveCfg = Release|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Release|x64.Build.0 = Release|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Release|x86.ActiveCfg = Release|Any CPU
|
||||
{7CF9ED88-1F32-4040-BEB1-D0902E304C70}.Release|x86.Build.0 = Release|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Debug|x64.ActiveCfg = Debug|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Debug|x64.Build.0 = Debug|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Debug|x86.ActiveCfg = Debug|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Debug|x86.Build.0 = Debug|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Release|x64.ActiveCfg = Release|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Release|x64.Build.0 = Release|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Release|x86.ActiveCfg = Release|Any CPU
|
||||
{9AB807A8-0469-40F7-A000-D240F36B6E5D}.Release|x86.Build.0 = Release|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Debug|x64.ActiveCfg = Debug|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Debug|x64.Build.0 = Debug|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Debug|x86.ActiveCfg = Debug|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Debug|x86.Build.0 = Debug|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Release|x64.ActiveCfg = Release|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Release|x64.Build.0 = Release|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Release|x86.ActiveCfg = Release|Any CPU
|
||||
{EB061E77-2475-4322-9257-3F2456DD141C}.Release|x86.Build.0 = Release|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Debug|Any CPU.Build.0 = Debug|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Debug|x64.ActiveCfg = Debug|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Debug|x64.Build.0 = Debug|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Debug|x86.ActiveCfg = Debug|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Debug|x86.Build.0 = Debug|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Release|Any CPU.ActiveCfg = Release|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Release|Any CPU.Build.0 = Release|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Release|x64.ActiveCfg = Release|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Release|x64.Build.0 = Release|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Release|x86.ActiveCfg = Release|Any CPU
|
||||
{B77B5A8E-0C53-4419-9BCD-227C9753A074}.Release|x86.Build.0 = Release|Any CPU
|
||||
EndGlobalSection
|
||||
GlobalSection(SolutionProperties) = preSolution
|
||||
HideSolutionNode = FALSE
|
||||
EndGlobalSection
|
||||
EndGlobal
|
||||
@@ -1,24 +0,0 @@
|
||||
namespace MxGateway.Client;
|
||||
|
||||
public sealed record DiscoverHierarchyOptions
|
||||
{
|
||||
public int? RootGobjectId { get; init; }
|
||||
|
||||
public string? RootTagName { get; init; }
|
||||
|
||||
public string? RootContainedPath { get; init; }
|
||||
|
||||
public int? MaxDepth { get; init; }
|
||||
|
||||
public IReadOnlyList<int> CategoryIds { get; init; } = Array.Empty<int>();
|
||||
|
||||
public IReadOnlyList<string> TemplateChainContains { get; init; } = Array.Empty<string>();
|
||||
|
||||
public string? TagNameGlob { get; init; }
|
||||
|
||||
public bool? IncludeAttributes { get; init; }
|
||||
|
||||
public bool AlarmBearingOnly { get; init; }
|
||||
|
||||
public bool HistorizedOnly { get; init; }
|
||||
}
|
||||
@@ -1,25 +0,0 @@
|
||||
using MxGateway.Contracts;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Exposes the protocol versions compiled into this client package.
|
||||
/// </summary>
|
||||
public static class MxGatewayClientContractInfo
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets the gateway gRPC protocol version compiled into this client package.
|
||||
/// A client and gateway are wire-compatible only when this value matches the
|
||||
/// gateway's advertised gateway protocol version.
|
||||
/// </summary>
|
||||
public const uint GatewayProtocolVersion =
|
||||
GatewayContractInfo.GatewayProtocolVersion;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the worker frame protocol version compiled into this client package.
|
||||
/// Exposed for diagnostics so callers can report the worker protocol the
|
||||
/// shared contracts were generated against.
|
||||
/// </summary>
|
||||
public const uint WorkerProtocolVersion =
|
||||
GatewayContractInfo.WorkerProtocolVersion;
|
||||
}
|
||||
@@ -1,3 +0,0 @@
|
||||
using System.Runtime.CompilerServices;
|
||||
|
||||
[assembly: InternalsVisibleTo("MxGateway.Client.Tests")]
|
||||
@@ -1,55 +0,0 @@
|
||||
using Grpc.Core;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Maps low-level <see cref="RpcException"/>s raised by the gRPC stack to the client's
|
||||
/// native exception hierarchy. Shared by every gateway and Galaxy Repository transport
|
||||
/// so the gRPC-to-native translation has exactly one implementation.
|
||||
/// </summary>
|
||||
internal static class RpcExceptionMapper
|
||||
{
|
||||
/// <summary>
|
||||
/// Translates a <see cref="RpcException"/> into the most specific native exception type.
|
||||
/// </summary>
|
||||
/// <param name="exception">The gRPC exception to translate.</param>
|
||||
/// <param name="cancellationToken">
|
||||
/// The cancellation token of the originating call; used to distinguish a caller-driven
|
||||
/// cancellation from a server-side <see cref="StatusCode.Cancelled"/> status.
|
||||
/// </param>
|
||||
/// <returns>
|
||||
/// An <see cref="OperationCanceledException"/> when the call was cancelled, a typed
|
||||
/// authentication/authorization exception for auth statuses, or an
|
||||
/// <see cref="MxGatewayException"/> carrying the originating gRPC <see cref="StatusCode"/>.
|
||||
/// </returns>
|
||||
public static Exception Map(
|
||||
RpcException exception,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(exception);
|
||||
|
||||
if (cancellationToken.IsCancellationRequested || exception.StatusCode == StatusCode.Cancelled)
|
||||
{
|
||||
return new OperationCanceledException(
|
||||
exception.Status.Detail,
|
||||
exception,
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
return exception.StatusCode switch
|
||||
{
|
||||
StatusCode.Unauthenticated => new MxGatewayAuthenticationException(
|
||||
exception.Status.Detail,
|
||||
statusCode: exception.StatusCode,
|
||||
innerException: exception),
|
||||
StatusCode.PermissionDenied => new MxGatewayAuthorizationException(
|
||||
exception.Status.Detail,
|
||||
statusCode: exception.StatusCode,
|
||||
innerException: exception),
|
||||
_ => new MxGatewayException(
|
||||
exception.Status.Detail,
|
||||
exception.StatusCode,
|
||||
exception),
|
||||
};
|
||||
}
|
||||
}
|
||||
+28
-40
@@ -7,11 +7,11 @@ CLI, and unit tests.
|
||||
|
||||
| Project | Purpose |
|
||||
|---------|---------|
|
||||
| `MxGateway.Client` | .NET 10 library entry point, raw gRPC calls, and session helpers. |
|
||||
| `MxGateway.Client.Cli` | Test CLI for smoke and diagnostic commands. |
|
||||
| `MxGateway.Client.Tests` | Unit tests for client options, generated contract wiring, auth metadata, session helpers, cancellation, and event streaming. |
|
||||
| `ZB.MOM.WW.MxGateway.Client` | .NET 10 library entry point, raw gRPC calls, and session helpers. |
|
||||
| `ZB.MOM.WW.MxGateway.Client.Cli` | Test CLI for smoke and diagnostic commands. |
|
||||
| `ZB.MOM.WW.MxGateway.Client.Tests` | Unit tests for client options, generated contract wiring, auth metadata, session helpers, cancellation, and event streaming. |
|
||||
|
||||
The projects reference `src/MxGateway.Contracts/MxGateway.Contracts.csproj` so
|
||||
The projects reference `src/ZB.MOM.WW.MxGateway.Contracts/ZB.MOM.WW.MxGateway.Contracts.csproj` so
|
||||
the client compiles against the same generated protobuf and gRPC types as the
|
||||
gateway. `clients/dotnet/generated` remains reserved for generator output if a
|
||||
future client build switches to client-local `Grpc.Tools` generation.
|
||||
@@ -19,8 +19,8 @@ future client build switches to client-local `Grpc.Tools` generation.
|
||||
## Build And Test
|
||||
|
||||
```powershell
|
||||
dotnet build clients/dotnet/MxGateway.Client.sln
|
||||
dotnet test clients/dotnet/MxGateway.Client.sln --no-build
|
||||
dotnet build clients/dotnet/ZB.MOM.WW.MxGateway.Client.slnx
|
||||
dotnet test clients/dotnet/ZB.MOM.WW.MxGateway.Client.slnx --no-build
|
||||
```
|
||||
|
||||
## Packaging
|
||||
@@ -29,8 +29,8 @@ Create local library and CLI artifacts from the repository root:
|
||||
|
||||
```powershell
|
||||
$dotnetPackageOutput = Join-Path (Get-Location) 'artifacts/clients/dotnet'
|
||||
dotnet pack clients/dotnet/MxGateway.Client/MxGateway.Client.csproj -c Release -p:PackageOutputPath="$dotnetPackageOutput"
|
||||
dotnet publish clients/dotnet/MxGateway.Client.Cli/MxGateway.Client.Cli.csproj -c Release -o artifacts/clients/dotnet/mxgw-dotnet
|
||||
dotnet pack clients/dotnet/ZB.MOM.WW.MxGateway.Client/ZB.MOM.WW.MxGateway.Client.csproj -c Release -p:PackageOutputPath="$dotnetPackageOutput"
|
||||
dotnet publish clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli/ZB.MOM.WW.MxGateway.Client.Cli.csproj -c Release -o artifacts/clients/dotnet/mxgw-dotnet
|
||||
```
|
||||
|
||||
The library package references the shared contracts project at build time. The
|
||||
@@ -39,11 +39,11 @@ published CLI runs from `artifacts/clients/dotnet/mxgw-dotnet`.
|
||||
## Regenerating Protobuf Bindings
|
||||
|
||||
The .NET client uses the generated C# types from
|
||||
`src/MxGateway.Contracts/Generated`. Regenerate those files through the
|
||||
`src/ZB.MOM.WW.MxGateway.Contracts/Generated`. Regenerate those files through the
|
||||
contracts project:
|
||||
|
||||
```powershell
|
||||
dotnet build src/MxGateway.Contracts/MxGateway.Contracts.csproj
|
||||
dotnet build src/ZB.MOM.WW.MxGateway.Contracts/ZB.MOM.WW.MxGateway.Contracts.csproj
|
||||
```
|
||||
|
||||
## Client Usage
|
||||
@@ -112,38 +112,26 @@ can keep the full `MxCommandReply`, HRESULT, and status array when MXAccess
|
||||
itself rejects a command. `MxAccessException.Reply` contains the raw generated
|
||||
reply.
|
||||
|
||||
When a gRPC call itself fails, the transport maps the underlying
|
||||
`RpcException` to a native exception: `Unauthenticated` becomes
|
||||
`MxGatewayAuthenticationException`, `PermissionDenied` becomes
|
||||
`MxGatewayAuthorizationException`, a cancelled call becomes
|
||||
`OperationCanceledException`, and every other status becomes a base
|
||||
`MxGatewayException`. `MxGatewayException.StatusCode` carries the originating
|
||||
gRPC `Grpc.Core.StatusCode` (non-null whenever the failure came from a gRPC
|
||||
status), so callers can distinguish a transient outage (`Unavailable`) from a
|
||||
permanent error (`InvalidArgument`, `NotFound`) without downcasting
|
||||
`InnerException`.
|
||||
|
||||
## CLI Usage
|
||||
|
||||
The test CLI supports deterministic JSON output for automation:
|
||||
|
||||
```powershell
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- version --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- open-session --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- register --session-id <id> --client-name mxgw-dotnet-cli --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- add-item --session-id <id> --server-handle 1 --item Area001.Pump001.Speed --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- advise --session-id <id> --server-handle 1 --item-handle 1 --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- write --session-id <id> --server-handle 1 --item-handle 1 --type int32 --value 123 --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- write2 --session-id <id> --server-handle 1 --item-handle 1 --type int32 --value 123 --timestamp 2026-01-01T00:00:00Z --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- stream-events --session-id <id> --max-events 1 --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- smoke --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --item Area001.Pump001.Speed --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- version --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- open-session --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- register --session-id <id> --client-name mxgw-dotnet-cli --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- add-item --session-id <id> --server-handle 1 --item Area001.Pump001.Speed --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- advise --session-id <id> --server-handle 1 --item-handle 1 --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- write --session-id <id> --server-handle 1 --item-handle 1 --type int32 --value 123 --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- write2 --session-id <id> --server-handle 1 --item-handle 1 --type int32 --value 123 --timestamp 2026-01-01T00:00:00Z --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- stream-events --session-id <id> --max-events 1 --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- smoke --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --item Area001.Pump001.Speed --json
|
||||
```
|
||||
|
||||
`smoke` opens a session, registers a client, adds one item, advises it,
|
||||
optionally writes a value when `--type` and `--value` are supplied, reads a
|
||||
bounded event stream, and closes the session in a `finally` block. CLI error
|
||||
output redacts the effective API key, whether it was supplied through
|
||||
`--api-key` or resolved from the `--api-key-env` environment variable.
|
||||
output redacts API keys supplied through `--api-key`.
|
||||
|
||||
## Galaxy Repository Browse
|
||||
|
||||
@@ -192,9 +180,9 @@ IReadOnlyList<GalaxyObject> pumps = await repository.DiscoverHierarchyAsync(
|
||||
The CLI exposes the same operations:
|
||||
|
||||
```powershell
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- galaxy-test-connection --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- galaxy-last-deploy --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- galaxy-discover --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- galaxy-test-connection --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- galaxy-last-deploy --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- galaxy-discover --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY
|
||||
```
|
||||
|
||||
### Watching deploy events
|
||||
@@ -229,15 +217,15 @@ await foreach (DeployEvent evt in repository.WatchDeployEventsAsync(
|
||||
The CLI counterpart streams events until Ctrl+C (or `--max-events`):
|
||||
|
||||
```powershell
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- galaxy-watch --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- galaxy-watch --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --last-seen-deploy-time 2026-04-28T14:30:00Z --json
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- galaxy-watch --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --max-events 5 --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- galaxy-watch --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- galaxy-watch --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --last-seen-deploy-time 2026-04-28T14:30:00Z --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- galaxy-watch --endpoint http://localhost:5000 --api-key-env MXGATEWAY_API_KEY --max-events 5 --json
|
||||
```
|
||||
|
||||
Use TLS options for a secured gateway:
|
||||
|
||||
```powershell
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- smoke --endpoint https://mxgateway.example.local:5001 --tls --ca-file C:\certs\mxgateway-ca.pem --server-name mxgateway.example.local --api-key-env MXGATEWAY_API_KEY --item Area001.Pump001.Speed --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- smoke --endpoint https://ZB.MOM.WW.MxGateway.example.local:5001 --tls --ca-file C:\certs\mxgateway-ca.pem --server-name ZB.MOM.WW.MxGateway.example.local --api-key-env MXGATEWAY_API_KEY --item Area001.Pump001.Speed --json
|
||||
```
|
||||
|
||||
## Integration Checks
|
||||
@@ -249,7 +237,7 @@ $env:MXGATEWAY_INTEGRATION = '1'
|
||||
$env:MXGATEWAY_ENDPOINT = 'http://localhost:5000'
|
||||
$env:MXGATEWAY_API_KEY = '<gateway-api-key>'
|
||||
$env:MXGATEWAY_TEST_ITEM = 'Area001.Pump001.Speed'
|
||||
dotnet run --project clients/dotnet/MxGateway.Client.Cli -- smoke --endpoint $env:MXGATEWAY_ENDPOINT --api-key-env MXGATEWAY_API_KEY --item $env:MXGATEWAY_TEST_ITEM --json
|
||||
dotnet run --project clients/dotnet/ZB.MOM.WW.MxGateway.Client.Cli -- smoke --endpoint $env:MXGATEWAY_ENDPOINT --api-key-env MXGATEWAY_API_KEY --item $env:MXGATEWAY_TEST_ITEM --json
|
||||
```
|
||||
|
||||
## Related Documentation
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
using System.Globalization;
|
||||
|
||||
namespace MxGateway.Client.Cli;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Cli;
|
||||
|
||||
/// <summary>Parses command-line arguments into flags and named values.</summary>
|
||||
internal sealed class CliArguments
|
||||
+3
-3
@@ -1,7 +1,7 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
|
||||
namespace MxGateway.Client.Cli;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Cli;
|
||||
|
||||
public interface IMxGatewayCliClient : IAsyncDisposable
|
||||
{
|
||||
+4
-4
@@ -1,8 +1,8 @@
|
||||
using MxGateway.Client;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Client;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
|
||||
namespace MxGateway.Client.Cli;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Cli;
|
||||
|
||||
internal sealed class MxGatewayCliClientAdapter : IMxGatewayCliClient
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MxGateway.Client.Cli;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Cli;
|
||||
|
||||
/// <summary>Utility to redact API keys from error messages for safe output.</summary>
|
||||
internal static class MxGatewayCliSecretRedactor
|
||||
+13
-30
@@ -1,11 +1,11 @@
|
||||
using System.Globalization;
|
||||
using System.Text.Json;
|
||||
using Google.Protobuf;
|
||||
using MxGateway.Client;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Client;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
|
||||
namespace MxGateway.Client.Cli;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Cli;
|
||||
|
||||
/// <summary>Command-line interface for the MXAccess Gateway client, supporting session and command operations.</summary>
|
||||
public static class MxGatewayClientCli
|
||||
@@ -122,10 +122,7 @@ public static class MxGatewayClientCli
|
||||
}
|
||||
catch (Exception exception) when (exception is not OperationCanceledException)
|
||||
{
|
||||
// Redact the effective API key — whether it came from --api-key or from
|
||||
// the (documented default) --api-key-env environment variable — so a
|
||||
// transport error message that echoes the bearer token is never printed.
|
||||
string? apiKey = TryResolveApiKey(arguments);
|
||||
string? apiKey = arguments.GetOptional("api-key");
|
||||
string message = MxGatewayCliSecretRedactor.Redact(exception.Message, apiKey);
|
||||
|
||||
if (arguments.HasFlag("json"))
|
||||
@@ -170,27 +167,6 @@ public static class MxGatewayClientCli
|
||||
}
|
||||
|
||||
private static string ResolveApiKey(CliArguments arguments)
|
||||
{
|
||||
string? apiKey = TryResolveApiKey(arguments);
|
||||
if (!string.IsNullOrWhiteSpace(apiKey))
|
||||
{
|
||||
return apiKey;
|
||||
}
|
||||
|
||||
string apiKeyEnvironmentName = arguments.GetOptional("api-key-env")
|
||||
?? "MXGATEWAY_API_KEY";
|
||||
|
||||
throw new ArgumentException(
|
||||
$"Gateway API key is required. Pass --api-key or set {apiKeyEnvironmentName}.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Resolves the effective API key from <c>--api-key</c> or, failing that, the
|
||||
/// environment variable named by <c>--api-key-env</c> (default
|
||||
/// <c>MXGATEWAY_API_KEY</c>). Returns <see langword="null"/> when no key is
|
||||
/// configured; used for redaction where a missing key must not throw.
|
||||
/// </summary>
|
||||
private static string? TryResolveApiKey(CliArguments arguments)
|
||||
{
|
||||
string? apiKey = arguments.GetOptional("api-key");
|
||||
if (!string.IsNullOrWhiteSpace(apiKey))
|
||||
@@ -201,7 +177,14 @@ public static class MxGatewayClientCli
|
||||
string apiKeyEnvironmentName = arguments.GetOptional("api-key-env")
|
||||
?? "MXGATEWAY_API_KEY";
|
||||
|
||||
return Environment.GetEnvironmentVariable(apiKeyEnvironmentName);
|
||||
apiKey = Environment.GetEnvironmentVariable(apiKeyEnvironmentName);
|
||||
if (!string.IsNullOrWhiteSpace(apiKey))
|
||||
{
|
||||
return apiKey;
|
||||
}
|
||||
|
||||
throw new ArgumentException(
|
||||
$"Gateway API key is required. Pass --api-key or set {apiKeyEnvironmentName}.");
|
||||
}
|
||||
|
||||
private static CancellationTokenSource CreateCancellation(CliArguments arguments, string command)
|
||||
+1
-1
@@ -1,3 +1,3 @@
|
||||
using MxGateway.Client.Cli;
|
||||
using ZB.MOM.WW.MxGateway.Client.Cli;
|
||||
|
||||
return await MxGatewayClientCli.RunAsync(args, Console.Out, Console.Error);
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\MxGateway.Client\MxGateway.Client.csproj" />
|
||||
<ProjectReference Include="..\ZB.MOM.WW.MxGateway.Client\ZB.MOM.WW.MxGateway.Client.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<PropertyGroup>
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Fake Galaxy Repository client transport for testing.
|
||||
+7
-41
@@ -1,7 +1,7 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Fake implementation of IMxGatewayClientTransport for testing.
|
||||
@@ -91,19 +91,6 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
|
||||
/// </summary>
|
||||
public Queue<Exception> CloseSessionExceptions { get; } = new();
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets a value indicating whether thrown <see cref="RpcException"/>s are mapped
|
||||
/// to <see cref="MxGatewayException"/> the way the production gRPC transport does. Lets
|
||||
/// retry tests exercise the wrapped-exception predicate branch that runs in production.
|
||||
/// </summary>
|
||||
public bool MapTransportExceptions { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets an optional hook awaited inside CloseSessionAsync after the call is
|
||||
/// recorded; lets tests pause a close mid-flight to observe concurrent dispose.
|
||||
/// </summary>
|
||||
public Func<Task>? CloseSessionHook { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets the queue of exceptions to throw from InvokeAsync.
|
||||
/// </summary>
|
||||
@@ -121,7 +108,7 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
|
||||
OpenSessionCalls.Add((request, callOptions));
|
||||
if (OpenSessionExceptions.TryDequeue(out Exception? exception))
|
||||
{
|
||||
throw Translate(exception, callOptions);
|
||||
throw exception;
|
||||
}
|
||||
|
||||
return Task.FromResult(OpenSessionReply);
|
||||
@@ -132,23 +119,17 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
|
||||
/// </summary>
|
||||
/// <param name="request">The CloseSessionRequest to process.</param>
|
||||
/// <param name="callOptions">Call options specifying RPC behavior.</param>
|
||||
public async Task<CloseSessionReply> CloseSessionAsync(
|
||||
public Task<CloseSessionReply> CloseSessionAsync(
|
||||
CloseSessionRequest request,
|
||||
CallOptions callOptions)
|
||||
{
|
||||
CloseSessionCalls.Add((request, callOptions));
|
||||
|
||||
if (CloseSessionHook is not null)
|
||||
{
|
||||
await CloseSessionHook().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
if (CloseSessionExceptions.TryDequeue(out Exception? exception))
|
||||
{
|
||||
throw Translate(exception, callOptions);
|
||||
throw exception;
|
||||
}
|
||||
|
||||
return CloseSessionReply;
|
||||
return Task.FromResult(CloseSessionReply);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -163,7 +144,7 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
|
||||
InvokeCalls.Add((request, callOptions));
|
||||
if (InvokeExceptions.TryDequeue(out Exception? exception))
|
||||
{
|
||||
throw Translate(exception, callOptions);
|
||||
throw exception;
|
||||
}
|
||||
|
||||
return Task.FromResult(_invokeReplies.Dequeue());
|
||||
@@ -223,7 +204,6 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
|
||||
? _acknowledgeReplies.Dequeue()
|
||||
: new AcknowledgeAlarmReply
|
||||
{
|
||||
SessionId = request.SessionId,
|
||||
CorrelationId = request.ClientCorrelationId,
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
Status = new MxStatusProxy { Success = 1, Category = MxStatusCategory.Ok },
|
||||
@@ -258,18 +238,4 @@ internal sealed class FakeGatewayTransport(MxGatewayClientOptions options) : IMx
|
||||
{
|
||||
_activeAlarmSnapshots.Add(snapshot);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Maps a queued exception the way the production gRPC transport does when
|
||||
/// <see cref="MapTransportExceptions"/> is set; otherwise returns it unchanged.
|
||||
/// </summary>
|
||||
private Exception Translate(Exception exception, CallOptions callOptions)
|
||||
{
|
||||
if (MapTransportExceptions && exception is RpcException rpcException)
|
||||
{
|
||||
return RpcExceptionMapper.Map(rpcException, callOptions.CancellationToken);
|
||||
}
|
||||
|
||||
return exception;
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -1,8 +1,8 @@
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
public sealed class GalaxyRepositoryClientTests
|
||||
{
|
||||
+3
-3
@@ -1,8 +1,8 @@
|
||||
using Google.Protobuf;
|
||||
using MxGateway.Client;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Client;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
public sealed class MxCommandReplyExtensionsTests
|
||||
{
|
||||
+2
-6
@@ -1,8 +1,8 @@
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// PR E.2 — pins the .NET SDK surface for the new alarm RPCs:
|
||||
@@ -17,7 +17,6 @@ public sealed class MxGatewayClientAlarmsTests
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.AddAcknowledgeReply(new AcknowledgeAlarmReply
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
CorrelationId = "corr-1",
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
Status = new MxStatusProxy
|
||||
@@ -31,7 +30,6 @@ public sealed class MxGatewayClientAlarmsTests
|
||||
|
||||
AcknowledgeAlarmReply reply = await client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
ClientCorrelationId = "corr-1",
|
||||
AlarmFullReference = "Tank01.Level.HiHi",
|
||||
Comment = "investigating",
|
||||
@@ -64,7 +62,6 @@ public sealed class MxGatewayClientAlarmsTests
|
||||
client.AcknowledgeAlarmAsync(
|
||||
new AcknowledgeAlarmRequest
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
AlarmFullReference = "Tank01.Level.HiHi",
|
||||
Comment = string.Empty,
|
||||
OperatorUser = "alice",
|
||||
@@ -89,7 +86,6 @@ public sealed class MxGatewayClientAlarmsTests
|
||||
var ex = await Assert.ThrowsAsync<RpcException>(
|
||||
() => client.AcknowledgeAlarmAsync(new AcknowledgeAlarmRequest
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
AlarmFullReference = "Tank01.Level.HiHi",
|
||||
Comment = string.Empty,
|
||||
OperatorUser = "alice",
|
||||
+4
-41
@@ -1,9 +1,9 @@
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using MxGateway.Client.Cli;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Client.Cli;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
/// <summary>Tests for the CLI command interface.</summary>
|
||||
public sealed class MxGatewayClientCliTests
|
||||
@@ -106,43 +106,6 @@ public sealed class MxGatewayClientCliTests
|
||||
Assert.Contains("[redacted]", error.ToString());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that error output redacts the API key even when it was sourced from
|
||||
/// the <c>--api-key-env</c> environment variable rather than passed via
|
||||
/// <c>--api-key</c> — the documented default credential path.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task RunAsync_ErrorOutput_RedactsApiKey_WhenSourcedFromEnvironmentVariable()
|
||||
{
|
||||
const string environmentVariableName = "MXGATEWAY_TEST_API_KEY_REDACT";
|
||||
using var output = new StringWriter();
|
||||
using var error = new StringWriter();
|
||||
|
||||
Environment.SetEnvironmentVariable(environmentVariableName, "env-secret-api-key");
|
||||
try
|
||||
{
|
||||
int exitCode = await MxGatewayClientCli.RunAsync(
|
||||
[
|
||||
"open-session",
|
||||
"--endpoint",
|
||||
"http://localhost:5000",
|
||||
"--api-key-env",
|
||||
environmentVariableName,
|
||||
],
|
||||
output,
|
||||
error,
|
||||
_ => throw new InvalidOperationException("boom env-secret-api-key"));
|
||||
|
||||
Assert.Equal(1, exitCode);
|
||||
Assert.DoesNotContain("env-secret-api-key", error.ToString());
|
||||
Assert.Contains("[redacted]", error.ToString());
|
||||
}
|
||||
finally
|
||||
{
|
||||
Environment.SetEnvironmentVariable(environmentVariableName, null);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Verifies that stream-events with max-events limit stops output in non-JSON format.</summary>
|
||||
[Fact]
|
||||
public async Task RunAsync_StreamEvents_WithMaxEventsStopsNonJsonOutput()
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MxGateway.Contracts;
|
||||
using ZB.MOM.WW.MxGateway.Contracts;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
public sealed class MxGatewayClientContractInfoTests
|
||||
{
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
public sealed class MxGatewayClientOptionsTests
|
||||
{
|
||||
+2
-155
@@ -1,7 +1,7 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using Grpc.Core;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
/// <summary>Tests for MxGatewaySession and client command behavior.</summary>
|
||||
public sealed class MxGatewayClientSessionTests
|
||||
@@ -231,52 +231,6 @@ public sealed class MxGatewayClientSessionTests
|
||||
Assert.Equal("session-fixture", call.Request.SessionId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that disposing a session while other callers are concurrently inside
|
||||
/// <see cref="MxGatewaySession.CloseAsync"/> — one holding the close lock and one
|
||||
/// parked on it — never throws <see cref="ObjectDisposedException"/> into those
|
||||
/// callers. The close lock must outlive every pending close.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task DisposeAsync_DoesNotRaceConcurrentCloseAsync()
|
||||
{
|
||||
for (int iteration = 0; iteration < 100; iteration++)
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
using SemaphoreSlim firstCloseEntered = new(0, 1);
|
||||
using SemaphoreSlim releaseFirstClose = new(0, 1);
|
||||
|
||||
// The first CloseAsync to reach the transport parks here while holding the
|
||||
// session's close lock; later callers queue on the lock behind it.
|
||||
transport.CloseSessionHook = async () =>
|
||||
{
|
||||
firstCloseEntered.Release();
|
||||
await releaseFirstClose.WaitAsync().ConfigureAwait(false);
|
||||
transport.CloseSessionHook = null;
|
||||
};
|
||||
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
MxGatewaySession session = await client.OpenSessionAsync();
|
||||
|
||||
// Holder enters CloseAsync, acquires the lock, and parks in the hook.
|
||||
Task holder = Task.Run(() => session.CloseAsync());
|
||||
await firstCloseEntered.WaitAsync();
|
||||
|
||||
// Waiter is parked on the close lock behind the holder.
|
||||
Task waiter = Task.Run(() => session.CloseAsync());
|
||||
|
||||
// DisposeAsync runs concurrently; it must wait out both callers before
|
||||
// disposing the close lock rather than tearing it down underneath them.
|
||||
Task dispose = session.DisposeAsync().AsTask();
|
||||
|
||||
releaseFirstClose.Release();
|
||||
|
||||
await holder;
|
||||
await waiter;
|
||||
await dispose;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Verifies that invoke retries safe diagnostic commands on transient RPC failure.</summary>
|
||||
[Fact]
|
||||
public async Task InvokeAsync_RetriesSafeDiagnosticCommandOnTransientGrpcFailure()
|
||||
@@ -301,35 +255,6 @@ public sealed class MxGatewayClientSessionTests
|
||||
Assert.Equal(2, transport.InvokeCalls.Count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that the retry pipeline still retries when the transport maps the raw
|
||||
/// <see cref="RpcException"/> to an <see cref="MxGatewayException"/> before it reaches
|
||||
/// the retry predicate — the wrapped-exception shape that production always produces.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task InvokeAsync_RetriesSafeDiagnosticCommand_WhenTransportMapsRpcException()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.MapTransportExceptions = true;
|
||||
transport.InvokeExceptions.Enqueue(CreateTransientRpcException());
|
||||
transport.AddInvokeReply(new MxCommandReply
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
Kind = MxCommandKind.Ping,
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
});
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
MxGatewaySession session = await client.OpenSessionAsync();
|
||||
|
||||
await session.InvokeAsync(new MxCommandRequest
|
||||
{
|
||||
SessionId = session.SessionId,
|
||||
Command = new MxCommand { Kind = MxCommandKind.Ping, Ping = new PingCommand() },
|
||||
});
|
||||
|
||||
Assert.Equal(2, transport.InvokeCalls.Count);
|
||||
}
|
||||
|
||||
/// <summary>Verifies that open session does not retry on transient RPC failure.</summary>
|
||||
[Fact]
|
||||
public async Task OpenSessionAsync_DoesNotRetryTransientGrpcFailure()
|
||||
@@ -378,84 +303,6 @@ public sealed class MxGatewayClientSessionTests
|
||||
Assert.Equal(cancellation.Token, Assert.Single(transport.InvokeCalls).CallOptions.CancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that a client-imposed <see cref="StatusCode.DeadlineExceeded"/> is not
|
||||
/// retried. The deadline budget is shared across the whole safe-unary operation, so
|
||||
/// an immediate retry would only fail again — the call must surface the failure.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task InvokeAsync_DoesNotRetrySafeDiagnosticCommand_OnDeadlineExceeded()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.InvokeExceptions.Enqueue(
|
||||
new RpcException(new Status(StatusCode.DeadlineExceeded, "deadline exceeded")));
|
||||
transport.AddInvokeReply(new MxCommandReply
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
Kind = MxCommandKind.Ping,
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
});
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
MxGatewaySession session = await client.OpenSessionAsync();
|
||||
|
||||
await Assert.ThrowsAsync<RpcException>(async () => await session.InvokeAsync(
|
||||
new MxCommandRequest
|
||||
{
|
||||
SessionId = session.SessionId,
|
||||
Command = new MxCommand { Kind = MxCommandKind.Ping, Ping = new PingCommand() },
|
||||
}));
|
||||
|
||||
Assert.Single(transport.InvokeCalls);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that a successful register reply missing the typed <c>register</c>
|
||||
/// payload throws a descriptive <see cref="MxGatewayException"/> rather than
|
||||
/// silently returning a zero server handle.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task RegisterAsync_Throws_WhenSuccessfulReplyMissingPayload()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.AddInvokeReply(new MxCommandReply
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
Kind = MxCommandKind.Register,
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
});
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
MxGatewaySession session = await client.OpenSessionAsync();
|
||||
|
||||
MxGatewayException exception = await Assert.ThrowsAsync<MxGatewayException>(
|
||||
async () => await session.RegisterAsync("client-name"));
|
||||
|
||||
Assert.Contains("register", exception.Message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that a successful add-item reply missing the typed <c>add_item</c>
|
||||
/// payload throws a descriptive <see cref="MxGatewayException"/> rather than
|
||||
/// silently returning a zero item handle.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task AddItemAsync_Throws_WhenSuccessfulReplyMissingPayload()
|
||||
{
|
||||
FakeGatewayTransport transport = CreateTransport();
|
||||
transport.AddInvokeReply(new MxCommandReply
|
||||
{
|
||||
SessionId = "session-fixture",
|
||||
Kind = MxCommandKind.AddItem,
|
||||
ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok },
|
||||
});
|
||||
await using MxGatewayClient client = CreateClient(transport);
|
||||
MxGatewaySession session = await client.OpenSessionAsync();
|
||||
|
||||
MxGatewayException exception = await Assert.ThrowsAsync<MxGatewayException>(
|
||||
async () => await session.AddItemAsync(1, "Area.Pump.Speed"));
|
||||
|
||||
Assert.Contains("add_item", exception.Message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
private static MxGatewayClient CreateClient(FakeGatewayTransport transport)
|
||||
{
|
||||
return new MxGatewayClient(transport.Options, transport);
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
public sealed class MxGatewayGeneratedContractTests
|
||||
{
|
||||
+3
-3
@@ -1,9 +1,9 @@
|
||||
using System.Text.Json;
|
||||
using Google.Protobuf;
|
||||
using MxGateway.Client;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Client;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
public sealed class MxStatusProxyExtensionsTests
|
||||
{
|
||||
+3
-3
@@ -1,9 +1,9 @@
|
||||
using System.Text.Json;
|
||||
using Google.Protobuf;
|
||||
using MxGateway.Client;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Client;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client.Tests;
|
||||
namespace ZB.MOM.WW.MxGateway.Client.Tests;
|
||||
|
||||
public sealed class MxValueExtensionsTests
|
||||
{
|
||||
+2
-2
@@ -19,8 +19,8 @@
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\MxGateway.Client\MxGateway.Client.csproj" />
|
||||
<ProjectReference Include="..\MxGateway.Client.Cli\MxGateway.Client.Cli.csproj" />
|
||||
<ProjectReference Include="..\ZB.MOM.WW.MxGateway.Client\ZB.MOM.WW.MxGateway.Client.csproj" />
|
||||
<ProjectReference Include="..\ZB.MOM.WW.MxGateway.Client.Cli\ZB.MOM.WW.MxGateway.Client.Cli.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
@@ -0,0 +1,11 @@
|
||||
<Solution>
|
||||
<Configurations>
|
||||
<Platform Name="Any CPU" />
|
||||
<Platform Name="x64" />
|
||||
<Platform Name="x86" />
|
||||
</Configurations>
|
||||
<Project Path="../../src/ZB.MOM.WW.MxGateway.Contracts/ZB.MOM.WW.MxGateway.Contracts.csproj" />
|
||||
<Project Path="ZB.MOM.WW.MxGateway.Client.Cli/ZB.MOM.WW.MxGateway.Client.Cli.csproj" />
|
||||
<Project Path="ZB.MOM.WW.MxGateway.Client.Tests/ZB.MOM.WW.MxGateway.Client.Tests.csproj" />
|
||||
<Project Path="ZB.MOM.WW.MxGateway.Client/ZB.MOM.WW.MxGateway.Client.csproj" />
|
||||
</Solution>
|
||||
@@ -0,0 +1,43 @@
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Filters and shape options for <see cref="GalaxyRepositoryClient.DiscoverHierarchyAsync(DiscoverHierarchyOptions, System.Threading.CancellationToken)"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Hand-written ergonomic wrapper around the generated
|
||||
/// <c>DiscoverHierarchyRequest</c>: lets callers express a Galaxy-browse
|
||||
/// slice with .NET-friendly nullable scalars and collection initializers,
|
||||
/// without touching the protobuf message's <c>oneof root</c> directly.
|
||||
/// </remarks>
|
||||
public sealed class DiscoverHierarchyOptions
|
||||
{
|
||||
/// <summary>Restrict to the subtree rooted at this Galaxy <c>gobject_id</c>.</summary>
|
||||
public int? RootGobjectId { get; init; }
|
||||
|
||||
/// <summary>Restrict to the subtree rooted at the object with this tag name.</summary>
|
||||
public string? RootTagName { get; init; }
|
||||
|
||||
/// <summary>Restrict to the subtree rooted at this <c>contained_name</c> path.</summary>
|
||||
public string? RootContainedPath { get; init; }
|
||||
|
||||
/// <summary>Maximum traversal depth, measured from the chosen root.</summary>
|
||||
public int? MaxDepth { get; init; }
|
||||
|
||||
/// <summary>Restrict to objects whose Galaxy category is in this set.</summary>
|
||||
public IReadOnlyList<int> CategoryIds { get; init; } = [];
|
||||
|
||||
/// <summary>Restrict to objects whose template chain contains any of these tokens.</summary>
|
||||
public IReadOnlyList<string> TemplateChainContains { get; init; } = [];
|
||||
|
||||
/// <summary>Optional glob-style filter on <c>tag_name</c>.</summary>
|
||||
public string? TagNameGlob { get; init; }
|
||||
|
||||
/// <summary>Whether to populate each <c>GalaxyObject.Attributes</c>. Null leaves the server default.</summary>
|
||||
public bool? IncludeAttributes { get; init; }
|
||||
|
||||
/// <summary>Restrict to objects that bear at least one alarm attribute.</summary>
|
||||
public bool AlarmBearingOnly { get; init; }
|
||||
|
||||
/// <summary>Restrict to objects that have at least one historized attribute.</summary>
|
||||
public bool HistorizedOnly { get; init; }
|
||||
}
|
||||
+2
-2
@@ -2,14 +2,14 @@ using Google.Protobuf.WellKnownTypes;
|
||||
using Grpc.Core;
|
||||
using Grpc.Net.Client;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
using Polly;
|
||||
using System.Net.Http;
|
||||
using System.Net.Security;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Provides the .NET client entry point for the public Galaxy Repository gRPC API.
|
||||
+30
-6
@@ -1,7 +1,7 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// gRPC implementation of IGalaxyRepositoryClientTransport.
|
||||
@@ -36,7 +36,7 @@ internal sealed class GrpcGalaxyRepositoryClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, callOptions.CancellationToken);
|
||||
throw MapRpcException(exception, callOptions.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -53,7 +53,7 @@ internal sealed class GrpcGalaxyRepositoryClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, callOptions.CancellationToken);
|
||||
throw MapRpcException(exception, callOptions.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -70,7 +70,7 @@ internal sealed class GrpcGalaxyRepositoryClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, callOptions.CancellationToken);
|
||||
throw MapRpcException(exception, callOptions.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -101,7 +101,7 @@ internal sealed class GrpcGalaxyRepositoryClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, effectiveCancellationToken);
|
||||
throw MapRpcException(exception, effectiveCancellationToken);
|
||||
}
|
||||
|
||||
yield return deployEvent;
|
||||
@@ -115,4 +115,28 @@ internal sealed class GrpcGalaxyRepositoryClientTransport(
|
||||
{
|
||||
return WatchDeployEventsAsync(request, callOptions);
|
||||
}
|
||||
|
||||
private static Exception MapRpcException(
|
||||
RpcException exception,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested || exception.StatusCode == StatusCode.Cancelled)
|
||||
{
|
||||
return new OperationCanceledException(
|
||||
exception.Status.Detail,
|
||||
exception,
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
return exception.StatusCode switch
|
||||
{
|
||||
StatusCode.Unauthenticated => new MxGatewayAuthenticationException(
|
||||
exception.Status.Detail,
|
||||
innerException: exception),
|
||||
StatusCode.PermissionDenied => new MxGatewayAuthorizationException(
|
||||
exception.Status.Detail,
|
||||
innerException: exception),
|
||||
_ => new MxGatewayException(exception.Status.Detail, exception),
|
||||
};
|
||||
}
|
||||
}
|
||||
+32
-8
@@ -1,7 +1,7 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// gRPC implementation of IMxGatewayClientTransport.
|
||||
@@ -36,7 +36,7 @@ internal sealed class GrpcMxGatewayClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, callOptions.CancellationToken);
|
||||
throw MapRpcException(exception, callOptions.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -53,7 +53,7 @@ internal sealed class GrpcMxGatewayClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, callOptions.CancellationToken);
|
||||
throw MapRpcException(exception, callOptions.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -70,7 +70,7 @@ internal sealed class GrpcMxGatewayClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, callOptions.CancellationToken);
|
||||
throw MapRpcException(exception, callOptions.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -101,7 +101,7 @@ internal sealed class GrpcMxGatewayClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, effectiveCancellationToken);
|
||||
throw MapRpcException(exception, effectiveCancellationToken);
|
||||
}
|
||||
|
||||
yield return gatewayEvent;
|
||||
@@ -129,7 +129,7 @@ internal sealed class GrpcMxGatewayClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, callOptions.CancellationToken);
|
||||
throw MapRpcException(exception, callOptions.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -160,7 +160,7 @@ internal sealed class GrpcMxGatewayClientTransport(
|
||||
}
|
||||
catch (RpcException exception)
|
||||
{
|
||||
throw RpcExceptionMapper.Map(exception, effectiveCancellationToken);
|
||||
throw MapRpcException(exception, effectiveCancellationToken);
|
||||
}
|
||||
|
||||
yield return snapshot;
|
||||
@@ -174,4 +174,28 @@ internal sealed class GrpcMxGatewayClientTransport(
|
||||
{
|
||||
return QueryActiveAlarmsAsync(request, callOptions);
|
||||
}
|
||||
|
||||
private static Exception MapRpcException(
|
||||
RpcException exception,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (cancellationToken.IsCancellationRequested || exception.StatusCode == StatusCode.Cancelled)
|
||||
{
|
||||
return new OperationCanceledException(
|
||||
exception.Status.Detail,
|
||||
exception,
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
return exception.StatusCode switch
|
||||
{
|
||||
StatusCode.Unauthenticated => new MxGatewayAuthenticationException(
|
||||
exception.Status.Detail,
|
||||
innerException: exception),
|
||||
StatusCode.PermissionDenied => new MxGatewayAuthorizationException(
|
||||
exception.Status.Detail,
|
||||
innerException: exception),
|
||||
_ => new MxGatewayException(exception.Status.Detail, exception),
|
||||
};
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto.Galaxy;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Transport layer for Galaxy Repository gRPC operations.</summary>
|
||||
internal interface IGalaxyRepositoryClientTransport
|
||||
+2
-2
@@ -1,7 +1,7 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
internal interface IMxGatewayClientTransport
|
||||
{
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Exception thrown when an MXAccess command fails with a non-zero HResult or failing status.</summary>
|
||||
public sealed class MxAccessException : MxGatewayCommandException
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Extension methods for checking MxCommandReply success conditions.</summary>
|
||||
public static class MxCommandReplyExtensions
|
||||
+4
-8
@@ -1,7 +1,6 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Exception thrown when an API key is invalid, expired, or malformed.</summary>
|
||||
public sealed class MxGatewayAuthenticationException : MxGatewayException
|
||||
@@ -14,7 +13,6 @@ public sealed class MxGatewayAuthenticationException : MxGatewayException
|
||||
/// <param name="hResult">The HResult code, if available.</param>
|
||||
/// <param name="statuses">The MXAccess statuses, if available.</param>
|
||||
/// <param name="innerException">The underlying exception, if any.</param>
|
||||
/// <param name="statusCode">The gRPC status code reported by the failed call, if available.</param>
|
||||
public MxGatewayAuthenticationException(
|
||||
string message,
|
||||
string? sessionId = null,
|
||||
@@ -22,8 +20,7 @@ public sealed class MxGatewayAuthenticationException : MxGatewayException
|
||||
ProtocolStatus? protocolStatus = null,
|
||||
int? hResult = null,
|
||||
IReadOnlyList<MxStatusProxy>? statuses = null,
|
||||
Exception? innerException = null,
|
||||
StatusCode? statusCode = null)
|
||||
Exception? innerException = null)
|
||||
: base(
|
||||
message,
|
||||
sessionId,
|
||||
@@ -31,8 +28,7 @@ public sealed class MxGatewayAuthenticationException : MxGatewayException
|
||||
protocolStatus,
|
||||
hResult,
|
||||
statuses ?? [],
|
||||
innerException,
|
||||
statusCode)
|
||||
innerException)
|
||||
{
|
||||
}
|
||||
}
|
||||
+4
-8
@@ -1,7 +1,6 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Exception thrown when the API key lacks required scopes for an operation.</summary>
|
||||
public sealed class MxGatewayAuthorizationException : MxGatewayException
|
||||
@@ -14,7 +13,6 @@ public sealed class MxGatewayAuthorizationException : MxGatewayException
|
||||
/// <param name="hResult">The HResult code, if available.</param>
|
||||
/// <param name="statuses">The MXAccess statuses, if available.</param>
|
||||
/// <param name="innerException">The underlying exception, if any.</param>
|
||||
/// <param name="statusCode">The gRPC status code reported by the failed call, if available.</param>
|
||||
public MxGatewayAuthorizationException(
|
||||
string message,
|
||||
string? sessionId = null,
|
||||
@@ -22,8 +20,7 @@ public sealed class MxGatewayAuthorizationException : MxGatewayException
|
||||
ProtocolStatus? protocolStatus = null,
|
||||
int? hResult = null,
|
||||
IReadOnlyList<MxStatusProxy>? statuses = null,
|
||||
Exception? innerException = null,
|
||||
StatusCode? statusCode = null)
|
||||
Exception? innerException = null)
|
||||
: base(
|
||||
message,
|
||||
sessionId,
|
||||
@@ -31,8 +28,7 @@ public sealed class MxGatewayAuthorizationException : MxGatewayException
|
||||
protocolStatus,
|
||||
hResult,
|
||||
statuses ?? [],
|
||||
innerException,
|
||||
statusCode)
|
||||
innerException)
|
||||
{
|
||||
}
|
||||
}
|
||||
+9
-9
@@ -1,13 +1,13 @@
|
||||
using Grpc.Core;
|
||||
using Grpc.Net.Client;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using Polly;
|
||||
using System.Net.Http;
|
||||
using System.Net.Security;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Provides the .NET client entry point for the public MXAccess Gateway gRPC API.
|
||||
@@ -17,7 +17,7 @@ public sealed class MxGatewayClient : IAsyncDisposable
|
||||
private readonly GrpcChannel _channel;
|
||||
private readonly IMxGatewayClientTransport _transport;
|
||||
private readonly ResiliencePipeline _safeUnaryRetryPipeline;
|
||||
private int _disposed;
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="MxGatewayClient"/> with given options and transport.
|
||||
@@ -184,10 +184,9 @@ public sealed class MxGatewayClient : IAsyncDisposable
|
||||
|
||||
/// <summary>
|
||||
/// Acknowledges an active MXAccess alarm condition through the gateway. The
|
||||
/// gateway authorizes <see cref="AcknowledgeAlarmRequest"/> against the API
|
||||
/// key's <c>admin</c> scope (there is no finer-grained alarm-ack sub-scope)
|
||||
/// and forwards the acknowledge to the worker's MXAccess session; the
|
||||
/// resulting <see cref="MxStatusProxy"/> is returned in the reply.
|
||||
/// gateway authenticates the request against the API key's <c>invoke:alarm-ack</c>
|
||||
/// scope and forwards the acknowledge to the worker's MXAccess session;
|
||||
/// the resulting <see cref="MxStatusProxy"/> is returned in the reply.
|
||||
/// </summary>
|
||||
/// <param name="request">The acknowledge request — alarm reference, comment, operator user.</param>
|
||||
/// <param name="cancellationToken">Cancellation token for the operation.</param>
|
||||
@@ -230,11 +229,12 @@ public sealed class MxGatewayClient : IAsyncDisposable
|
||||
/// </summary>
|
||||
public ValueTask DisposeAsync()
|
||||
{
|
||||
if (Interlocked.Exchange(ref _disposed, 1) != 0)
|
||||
if (_disposed)
|
||||
{
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
_channel?.Dispose();
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
@@ -335,6 +335,6 @@ public sealed class MxGatewayClient : IAsyncDisposable
|
||||
|
||||
private void ThrowIfDisposed()
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this);
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
using ZB.MOM.WW.MxGateway.Contracts;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Exposes the protocol versions compiled into this client package.
|
||||
/// </summary>
|
||||
public static class MxGatewayClientContractInfo
|
||||
{
|
||||
public const uint GatewayProtocolVersion =
|
||||
GatewayContractInfo.GatewayProtocolVersion;
|
||||
|
||||
public const uint WorkerProtocolVersion =
|
||||
GatewayContractInfo.WorkerProtocolVersion;
|
||||
}
|
||||
+2
-12
@@ -1,6 +1,6 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Configures the gRPC channel used by the .NET MXAccess Gateway client.
|
||||
@@ -38,12 +38,7 @@ public sealed class MxGatewayClientOptions
|
||||
public TimeSpan ConnectTimeout { get; init; } = TimeSpan.FromSeconds(10);
|
||||
|
||||
/// <summary>
|
||||
/// Gets the timeout budget for a unary gRPC operation. This is both the gRPC
|
||||
/// deadline stamped on each individual attempt and the overall budget for the
|
||||
/// whole safe-unary operation: for retryable calls the initial attempt, every
|
||||
/// retry, and the backoff delays between them all share this single budget.
|
||||
/// It is therefore an upper bound on the total wall-clock time a safe-unary
|
||||
/// call can take, not a fresh per-retry allowance.
|
||||
/// Gets the default timeout for unary gRPC calls.
|
||||
/// </summary>
|
||||
public TimeSpan DefaultCallTimeout { get; init; } = TimeSpan.FromSeconds(30);
|
||||
|
||||
@@ -52,11 +47,6 @@ public sealed class MxGatewayClientOptions
|
||||
/// </summary>
|
||||
public TimeSpan? StreamTimeout { get; init; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets the maximum size, in bytes, of a single gRPC message the client will
|
||||
/// send or receive. Applied to both the send and receive limits of the
|
||||
/// underlying channel. Defaults to 16 MiB.
|
||||
/// </summary>
|
||||
public int MaxGrpcMessageBytes { get; init; } = 16 * 1024 * 1024;
|
||||
|
||||
/// <summary>
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Configuration for automatic retry behavior on transient gRPC call failures.</summary>
|
||||
public sealed class MxGatewayClientRetryOptions
|
||||
+3
-8
@@ -1,10 +1,10 @@
|
||||
using Grpc.Core;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
using Polly;
|
||||
using Polly.Retry;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Factory and helpers for exponential-backoff retry policies on transient gRPC failures.</summary>
|
||||
internal static class MxGatewayClientRetryPolicy
|
||||
@@ -61,13 +61,8 @@ internal static class MxGatewayClientRetryPolicy
|
||||
|
||||
private static bool IsTransientStatus(StatusCode statusCode)
|
||||
{
|
||||
// DeadlineExceeded is intentionally NOT treated as transient. The deadline
|
||||
// on every unary call is client-imposed (CreateCallOptions stamps the
|
||||
// DefaultCallTimeout budget), and that same budget is shared across the
|
||||
// initial attempt plus all retries plus backoff. A DeadlineExceeded means
|
||||
// the shared budget is exhausted, so an immediate retry would only fail
|
||||
// again — burning the remaining budget on a call that cannot succeed.
|
||||
return statusCode is StatusCode.Unavailable
|
||||
or StatusCode.DeadlineExceeded
|
||||
or StatusCode.ResourceExhausted;
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Exception thrown when a gateway command fails due to an unclassified protocol error.</summary>
|
||||
public class MxGatewayCommandException : MxGatewayException
|
||||
+3
-32
@@ -1,7 +1,6 @@
|
||||
using Grpc.Core;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Exception thrown when a gateway RPC call fails or returns an error status.
|
||||
@@ -29,20 +28,6 @@ public class MxGatewayException : Exception
|
||||
Statuses = [];
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the MxGatewayException class carrying the originating
|
||||
/// gRPC status code so callers can distinguish transient from permanent failures.
|
||||
/// </summary>
|
||||
/// <param name="message">Diagnostic message describing the failure.</param>
|
||||
/// <param name="statusCode">The gRPC status code reported by the failed call.</param>
|
||||
/// <param name="innerException">Underlying exception that caused this failure.</param>
|
||||
public MxGatewayException(string message, StatusCode statusCode, Exception? innerException)
|
||||
: base(message, innerException)
|
||||
{
|
||||
StatusCode = statusCode;
|
||||
Statuses = [];
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the MxGatewayException class with full diagnostic information.
|
||||
/// </summary>
|
||||
@@ -53,7 +38,6 @@ public class MxGatewayException : Exception
|
||||
/// <param name="hResult">HRESULT code returned by the worker or MXAccess, if available.</param>
|
||||
/// <param name="statuses">List of MXAccess status codes returned by the operation.</param>
|
||||
/// <param name="innerException">Underlying exception that caused this failure.</param>
|
||||
/// <param name="statusCode">The gRPC status code reported by the failed call, if available.</param>
|
||||
public MxGatewayException(
|
||||
string message,
|
||||
string? sessionId,
|
||||
@@ -61,8 +45,7 @@ public class MxGatewayException : Exception
|
||||
ProtocolStatus? protocolStatus,
|
||||
int? hResult,
|
||||
IReadOnlyList<MxStatusProxy> statuses,
|
||||
Exception? innerException = null,
|
||||
StatusCode? statusCode = null)
|
||||
Exception? innerException = null)
|
||||
: base(message, innerException)
|
||||
{
|
||||
SessionId = sessionId;
|
||||
@@ -70,7 +53,6 @@ public class MxGatewayException : Exception
|
||||
ProtocolStatus = protocolStatus;
|
||||
HResultCode = hResult;
|
||||
Statuses = statuses;
|
||||
StatusCode = statusCode;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -97,15 +79,4 @@ public class MxGatewayException : Exception
|
||||
/// Gets the list of MXAccess status codes returned by the operation.
|
||||
/// </summary>
|
||||
public IReadOnlyList<MxStatusProxy> Statuses { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets the gRPC status code reported by the failed call, if the failure originated
|
||||
/// from a gRPC <see cref="RpcException"/>. <see langword="null"/> when the exception
|
||||
/// was not produced from a gRPC status (for example, a protocol-level reply failure).
|
||||
/// Callers can inspect this to distinguish a transient outage
|
||||
/// (<see cref="Grpc.Core.StatusCode.Unavailable"/>) from a permanent error
|
||||
/// (<see cref="Grpc.Core.StatusCode.InvalidArgument"/>) without downcasting
|
||||
/// <see cref="Exception.InnerException"/>.
|
||||
/// </summary>
|
||||
public StatusCode? StatusCode { get; }
|
||||
}
|
||||
+14
-81
@@ -1,6 +1,6 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Represents one gateway-backed MXAccess session.
|
||||
@@ -9,10 +9,7 @@ public sealed class MxGatewaySession : IAsyncDisposable
|
||||
{
|
||||
private readonly MxGatewayClient _client;
|
||||
private readonly SemaphoreSlim _closeLock = new(1, 1);
|
||||
private readonly object _disposeGate = new();
|
||||
private CloseSessionReply? _closeReply;
|
||||
private int _activeCloseCount;
|
||||
private bool _closeLockDisposed;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new session backed by the given MXAccess gateway client.
|
||||
@@ -49,42 +46,23 @@ public sealed class MxGatewaySession : IAsyncDisposable
|
||||
return _closeReply;
|
||||
}
|
||||
|
||||
// Register as an in-flight closer under the dispose gate. DisposeAsync waits for
|
||||
// _activeCloseCount to drain before disposing the close lock, so the semaphore is
|
||||
// guaranteed to outlive every WaitAsync started here.
|
||||
lock (_disposeGate)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_closeLockDisposed, this);
|
||||
_activeCloseCount++;
|
||||
}
|
||||
|
||||
await _closeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
await _closeLock.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
if (_closeReply is not null)
|
||||
{
|
||||
if (_closeReply is not null)
|
||||
{
|
||||
return _closeReply;
|
||||
}
|
||||
|
||||
_closeReply = await _client.CloseSessionRawAsync(
|
||||
new CloseSessionRequest { SessionId = SessionId },
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
return _closeReply;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_closeLock.Release();
|
||||
}
|
||||
|
||||
_closeReply = await _client.CloseSessionRawAsync(
|
||||
new CloseSessionRequest { SessionId = SessionId },
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
return _closeReply;
|
||||
}
|
||||
finally
|
||||
{
|
||||
lock (_disposeGate)
|
||||
{
|
||||
_activeCloseCount--;
|
||||
}
|
||||
_closeLock.Release();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -101,8 +79,7 @@ public sealed class MxGatewaySession : IAsyncDisposable
|
||||
MxCommandReply reply = await RegisterRawAsync(clientName, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
|
||||
return reply.Register?.ServerHandle
|
||||
?? throw CreateMissingPayloadException(reply, "register");
|
||||
return reply.Register?.ServerHandle ?? reply.ReturnValue.Int32Value;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -144,8 +121,7 @@ public sealed class MxGatewaySession : IAsyncDisposable
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
|
||||
return reply.AddItem?.ItemHandle
|
||||
?? throw CreateMissingPayloadException(reply, "add_item");
|
||||
return reply.AddItem?.ItemHandle ?? reply.ReturnValue.Int32Value;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -196,8 +172,7 @@ public sealed class MxGatewaySession : IAsyncDisposable
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
reply.EnsureProtocolSuccess().EnsureMxAccessSuccess();
|
||||
return reply.AddItem2?.ItemHandle
|
||||
?? throw CreateMissingPayloadException(reply, "add_item2");
|
||||
return reply.AddItem2?.ItemHandle ?? reply.ReturnValue.Int32Value;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -683,32 +658,7 @@ public sealed class MxGatewaySession : IAsyncDisposable
|
||||
/// </summary>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
lock (_disposeGate)
|
||||
{
|
||||
if (_closeLockDisposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
await CloseAsync().ConfigureAwait(false);
|
||||
|
||||
// Wait for every concurrent CloseAsync caller to leave the close lock before
|
||||
// disposing it; once _closeReply is set those callers return without awaiting.
|
||||
while (true)
|
||||
{
|
||||
lock (_disposeGate)
|
||||
{
|
||||
if (_activeCloseCount == 0)
|
||||
{
|
||||
_closeLockDisposed = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
await Task.Yield();
|
||||
}
|
||||
|
||||
_closeLock.Dispose();
|
||||
}
|
||||
|
||||
@@ -726,21 +676,4 @@ public sealed class MxGatewaySession : IAsyncDisposable
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds the exception thrown when a command reply passed protocol and
|
||||
/// MXAccess success checks but is missing the typed handle-bearing payload
|
||||
/// the command contract requires. Surfacing this as a clear error avoids
|
||||
/// silently handing a zero handle to the caller (it would otherwise fall
|
||||
/// through to <see cref="MxCommandReply.ReturnValue"/>, which is 0 when the
|
||||
/// reply carries no return value).
|
||||
/// </summary>
|
||||
private static MxGatewayException CreateMissingPayloadException(
|
||||
MxCommandReply reply,
|
||||
string expectedPayload)
|
||||
{
|
||||
return new MxGatewayException(
|
||||
$"Gateway reply for command kind={reply.Kind} reported success but is missing "
|
||||
+ $"the required '{expectedPayload}' payload; cannot resolve a handle. "
|
||||
+ $"session={reply.SessionId}; correlation={reply.CorrelationId}");
|
||||
}
|
||||
}
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Exception thrown when a session is not found, not ready, or invalid.</summary>
|
||||
public sealed class MxGatewaySessionException : MxGatewayException
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Exception thrown when the worker process is unavailable or fails to process a command.</summary>
|
||||
public sealed class MxGatewayWorkerException : MxGatewayException
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>Extension methods for MxStatusProxy values.</summary>
|
||||
public static class MxStatusProxyExtensions
|
||||
+2
-2
@@ -1,8 +1,8 @@
|
||||
using Google.Protobuf;
|
||||
using Google.Protobuf.WellKnownTypes;
|
||||
using MxGateway.Contracts.Proto;
|
||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||
|
||||
namespace MxGateway.Client;
|
||||
namespace ZB.MOM.WW.MxGateway.Client;
|
||||
|
||||
/// <summary>
|
||||
/// Creates and projects gateway MXAccess values without hiding the raw
|
||||
@@ -0,0 +1,3 @@
|
||||
using System.Runtime.CompilerServices;
|
||||
|
||||
[assembly: InternalsVisibleTo("ZB.MOM.WW.MxGateway.Client.Tests")]
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\src\MxGateway.Contracts\MxGateway.Contracts.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\ZB.MOM.WW.MxGateway.Contracts\ZB.MOM.WW.MxGateway.Contracts.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
+1
-20
@@ -79,30 +79,11 @@ client, err := mxgateway.Dial(ctx, mxgateway.Options{
|
||||
`AddItem`, `AddItem2`, `Advise`, `Write`, `Events`, and `Close`. Prefer
|
||||
`SubscribeEvents` or `SubscribeEventsAfter` for long-running streams because the
|
||||
returned subscription owns cancellation and exposes `Close` for deterministic
|
||||
goroutine cleanup. `Events` and `EventsAfter` are a compatibility shim with a
|
||||
bounded internal buffer: if the consumer drains too slowly the buffer fills,
|
||||
the underlying stream is cancelled, and a terminal `EventResult` carrying
|
||||
`ErrEventBufferOverflow` is delivered as the channel's last item before it
|
||||
closes — so a slow consumer can distinguish dropped events from a normal
|
||||
end-of-stream. `SubscribeEvents` blocks instead of dropping, so use it when no
|
||||
events may be lost. Raw protobuf messages remain available through the
|
||||
goroutine cleanup. Raw protobuf messages remain available through the
|
||||
`mxgateway` package aliases and the `Raw` helper methods. Typed errors support
|
||||
`errors.As` for `GatewayError`, `CommandError`, and `MxAccessError`; command
|
||||
errors preserve the raw reply.
|
||||
|
||||
`Dial` and `DialGalaxy` create the connection lazily (`grpc.NewClient`): a
|
||||
gateway that is briefly unavailable no longer turns into a hard error — the
|
||||
connection recovers once the gateway comes up. To keep fail-fast behavior,
|
||||
both run a readiness probe bounded by `DialTimeout` (default 10s, or the
|
||||
context deadline when sooner) and return a `*GatewayError` if the gateway
|
||||
cannot be reached in that window.
|
||||
|
||||
For retry, timeout, and auth handling, `GatewayError.Code()` exposes the
|
||||
wrapped gRPC `codes.Code`, and `mxgateway.IsTransient(err)` reports whether a
|
||||
failure (`Unavailable`, `DeadlineExceeded`, `ResourceExhausted`, `Aborted`)
|
||||
may succeed on retry — so callers do not have to unwrap the error and call
|
||||
`status.Code` themselves.
|
||||
|
||||
## Galaxy Repository browse
|
||||
|
||||
The `GalaxyRepository` service (proto package `galaxy_repository.v1`) is a
|
||||
|
||||
@@ -331,11 +331,6 @@ func runUnsubscribeBulk(ctx context.Context, args []string, stdout, stderr io.Wr
|
||||
return errors.New("session-id and item-handles are required")
|
||||
}
|
||||
|
||||
handles, err := parseInt32List(*itemHandles)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
client, options, err := dialForCommand(ctx, common)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -343,7 +338,7 @@ func runUnsubscribeBulk(ctx context.Context, args []string, stdout, stderr io.Wr
|
||||
defer client.Close()
|
||||
|
||||
session := mxgateway.NewSessionForID(client, *sessionID)
|
||||
results, err := session.UnsubscribeBulk(ctx, int32(*serverHandle), handles)
|
||||
results, err := session.UnsubscribeBulk(ctx, int32(*serverHandle), parseInt32List(*itemHandles))
|
||||
return writeBulkOutput(stdout, *jsonOutput, "unsubscribe-bulk", options, results, err)
|
||||
}
|
||||
|
||||
@@ -519,7 +514,7 @@ func parseStringList(value string) []string {
|
||||
return items
|
||||
}
|
||||
|
||||
func parseInt32List(value string) ([]int32, error) {
|
||||
func parseInt32List(value string) []int32 {
|
||||
parts := strings.Split(value, ",")
|
||||
items := make([]int32, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
@@ -529,11 +524,11 @@ func parseInt32List(value string) ([]int32, error) {
|
||||
}
|
||||
parsed, err := strconv.ParseInt(item, 10, 32)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid item handle %q: %w", item, err)
|
||||
panic(err)
|
||||
}
|
||||
items = append(items, int32(parsed))
|
||||
}
|
||||
return items, nil
|
||||
return items
|
||||
}
|
||||
|
||||
func bindCommonFlags(flags *flag.FlagSet) *commonOptions {
|
||||
|
||||
@@ -56,32 +56,3 @@ func TestParseValueBuildsTypedValue(t *testing.T) {
|
||||
t.Fatalf("int32 value = %d, want 123", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseInt32ListParsesValidTokens(t *testing.T) {
|
||||
items, err := parseInt32List("1, 2 ,3")
|
||||
if err != nil {
|
||||
t.Fatalf("parseInt32List() error = %v", err)
|
||||
}
|
||||
want := []int32{1, 2, 3}
|
||||
if len(items) != len(want) {
|
||||
t.Fatalf("parseInt32List() = %v, want %v", items, want)
|
||||
}
|
||||
for i := range want {
|
||||
if items[i] != want[i] {
|
||||
t.Fatalf("parseInt32List()[%d] = %d, want %d", i, items[i], want[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseInt32ListReturnsErrorOnMalformedToken(t *testing.T) {
|
||||
items, err := parseInt32List("1,foo")
|
||||
if err == nil {
|
||||
t.Fatalf("parseInt32List() error = nil, want a parse error; items = %v", items)
|
||||
}
|
||||
if items != nil {
|
||||
t.Fatalf("parseInt32List() items = %v, want nil on error", items)
|
||||
}
|
||||
if !strings.Contains(err.Error(), "foo") {
|
||||
t.Fatalf("parseInt32List() error = %q, want it to name the bad token", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@ Set-StrictMode -Version Latest
|
||||
$ErrorActionPreference = 'Stop'
|
||||
|
||||
$repoRoot = Resolve-Path (Join-Path $PSScriptRoot '..\..')
|
||||
$protoRoot = Join-Path $repoRoot 'src\MxGateway.Contracts\Protos'
|
||||
$protoRoot = Join-Path $repoRoot 'src\ZB.MOM.WW.MxGateway.Contracts\Protos'
|
||||
$outputRoot = Join-Path $PSScriptRoot 'internal\generated'
|
||||
$modulePath = 'gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated'
|
||||
$protoc = 'C:\Users\dohertj2\AppData\Local\Microsoft\WinGet\Packages\Google.Protobuf_Microsoft.Winget.Source_8wekyb3d8bbwe\bin\protoc.exe'
|
||||
|
||||
@@ -687,18 +687,32 @@ func (x *GalaxyObject) GetAttributes() []*GalaxyAttribute {
|
||||
}
|
||||
|
||||
type GalaxyAttribute struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
AttributeName string `protobuf:"bytes,1,opt,name=attribute_name,json=attributeName,proto3" json:"attribute_name,omitempty"`
|
||||
FullTagReference string `protobuf:"bytes,2,opt,name=full_tag_reference,json=fullTagReference,proto3" json:"full_tag_reference,omitempty"`
|
||||
MxDataType int32 `protobuf:"varint,3,opt,name=mx_data_type,json=mxDataType,proto3" json:"mx_data_type,omitempty"`
|
||||
DataTypeName string `protobuf:"bytes,4,opt,name=data_type_name,json=dataTypeName,proto3" json:"data_type_name,omitempty"`
|
||||
IsArray bool `protobuf:"varint,5,opt,name=is_array,json=isArray,proto3" json:"is_array,omitempty"`
|
||||
ArrayDimension int32 `protobuf:"varint,6,opt,name=array_dimension,json=arrayDimension,proto3" json:"array_dimension,omitempty"`
|
||||
ArrayDimensionPresent bool `protobuf:"varint,7,opt,name=array_dimension_present,json=arrayDimensionPresent,proto3" json:"array_dimension_present,omitempty"`
|
||||
MxAttributeCategory int32 `protobuf:"varint,8,opt,name=mx_attribute_category,json=mxAttributeCategory,proto3" json:"mx_attribute_category,omitempty"`
|
||||
SecurityClassification int32 `protobuf:"varint,9,opt,name=security_classification,json=securityClassification,proto3" json:"security_classification,omitempty"`
|
||||
IsHistorized bool `protobuf:"varint,10,opt,name=is_historized,json=isHistorized,proto3" json:"is_historized,omitempty"`
|
||||
IsAlarm bool `protobuf:"varint,11,opt,name=is_alarm,json=isAlarm,proto3" json:"is_alarm,omitempty"`
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
AttributeName string `protobuf:"bytes,1,opt,name=attribute_name,json=attributeName,proto3" json:"attribute_name,omitempty"`
|
||||
FullTagReference string `protobuf:"bytes,2,opt,name=full_tag_reference,json=fullTagReference,proto3" json:"full_tag_reference,omitempty"`
|
||||
// Raw Galaxy SQL `dbo.data_type` identifier, passed through unchanged.
|
||||
// This is NOT a member of `mxaccess_gateway.v1.MxDataType` — Galaxy's
|
||||
// type enumeration is distinct from MXAccess's wire data-type enum and
|
||||
// the two must not be cast or compared. The GalaxyRepository service is
|
||||
// metadata-only and deliberately does not share types with
|
||||
// mxaccess_gateway.proto. See docs/GalaxyRepository.md.
|
||||
MxDataType int32 `protobuf:"varint,3,opt,name=mx_data_type,json=mxDataType,proto3" json:"mx_data_type,omitempty"`
|
||||
// Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
// "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
DataTypeName string `protobuf:"bytes,4,opt,name=data_type_name,json=dataTypeName,proto3" json:"data_type_name,omitempty"`
|
||||
IsArray bool `protobuf:"varint,5,opt,name=is_array,json=isArray,proto3" json:"is_array,omitempty"`
|
||||
ArrayDimension int32 `protobuf:"varint,6,opt,name=array_dimension,json=arrayDimension,proto3" json:"array_dimension,omitempty"`
|
||||
ArrayDimensionPresent bool `protobuf:"varint,7,opt,name=array_dimension_present,json=arrayDimensionPresent,proto3" json:"array_dimension_present,omitempty"`
|
||||
// Raw Galaxy SQL attribute-category identifier, passed through unchanged.
|
||||
// Galaxy-specific; not mapped to any gateway enum. See
|
||||
// docs/GalaxyRepository.md.
|
||||
MxAttributeCategory int32 `protobuf:"varint,8,opt,name=mx_attribute_category,json=mxAttributeCategory,proto3" json:"mx_attribute_category,omitempty"`
|
||||
// Raw Galaxy SQL security-classification identifier, passed through
|
||||
// unchanged. Galaxy-specific; not mapped to any gateway enum. See
|
||||
// docs/GalaxyRepository.md.
|
||||
SecurityClassification int32 `protobuf:"varint,9,opt,name=security_classification,json=securityClassification,proto3" json:"security_classification,omitempty"`
|
||||
IsHistorized bool `protobuf:"varint,10,opt,name=is_historized,json=isHistorized,proto3" json:"is_historized,omitempty"`
|
||||
IsAlarm bool `protobuf:"varint,11,opt,name=is_alarm,json=isAlarm,proto3" json:"is_alarm,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -888,7 +902,7 @@ const file_galaxy_repository_proto_rawDesc = "" +
|
||||
"\x0eTestConnection\x12+.galaxy_repository.v1.TestConnectionRequest\x1a).galaxy_repository.v1.TestConnectionReply\x12q\n" +
|
||||
"\x11GetLastDeployTime\x12..galaxy_repository.v1.GetLastDeployTimeRequest\x1a,.galaxy_repository.v1.GetLastDeployTimeReply\x12q\n" +
|
||||
"\x11DiscoverHierarchy\x12..galaxy_repository.v1.DiscoverHierarchyRequest\x1a,.galaxy_repository.v1.DiscoverHierarchyReply\x12h\n" +
|
||||
"\x11WatchDeployEvents\x12..galaxy_repository.v1.WatchDeployEventsRequest\x1a!.galaxy_repository.v1.DeployEvent0\x01B#\xaa\x02 MxGateway.Contracts.Proto.Galaxyb\x06proto3"
|
||||
"\x11WatchDeployEvents\x12..galaxy_repository.v1.WatchDeployEventsRequest\x1a!.galaxy_repository.v1.DeployEvent0\x01B-\xaa\x02*ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxyb\x06proto3"
|
||||
|
||||
var (
|
||||
file_galaxy_repository_proto_rawDescOnce sync.Once
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -24,6 +24,7 @@ const (
|
||||
MxAccessGateway_Invoke_FullMethodName = "/mxaccess_gateway.v1.MxAccessGateway/Invoke"
|
||||
MxAccessGateway_StreamEvents_FullMethodName = "/mxaccess_gateway.v1.MxAccessGateway/StreamEvents"
|
||||
MxAccessGateway_AcknowledgeAlarm_FullMethodName = "/mxaccess_gateway.v1.MxAccessGateway/AcknowledgeAlarm"
|
||||
MxAccessGateway_StreamAlarms_FullMethodName = "/mxaccess_gateway.v1.MxAccessGateway/StreamAlarms"
|
||||
MxAccessGateway_QueryActiveAlarms_FullMethodName = "/mxaccess_gateway.v1.MxAccessGateway/QueryActiveAlarms"
|
||||
)
|
||||
|
||||
@@ -38,6 +39,17 @@ type MxAccessGatewayClient interface {
|
||||
Invoke(ctx context.Context, in *MxCommandRequest, opts ...grpc.CallOption) (*MxCommandReply, error)
|
||||
StreamEvents(ctx context.Context, in *StreamEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[MxEvent], error)
|
||||
AcknowledgeAlarm(ctx context.Context, in *AcknowledgeAlarmRequest, opts ...grpc.CallOption) (*AcknowledgeAlarmReply, error)
|
||||
// Session-less central alarm feed. The stream opens with the current
|
||||
// active-alarm snapshot (one `active_alarm` per alarm), then a single
|
||||
// `snapshot_complete`, then a `transition` for every subsequent change.
|
||||
// Served by the gateway's always-on alarm monitor; any number of clients
|
||||
// fan out from the single monitor without opening a worker session.
|
||||
StreamAlarms(ctx context.Context, in *StreamAlarmsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[AlarmFeedMessage], error)
|
||||
// Point-in-time snapshot of the currently-active alarm set served from the
|
||||
// gateway's always-on alarm monitor cache (session-less). Used after a
|
||||
// reconnect to seed Part 9 client state, or to reconcile alarms that may
|
||||
// have been missed during a transport blip. Streamed so callers can
|
||||
// begin processing without buffering the full set.
|
||||
QueryActiveAlarms(ctx context.Context, in *QueryActiveAlarmsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ActiveAlarmSnapshot], error)
|
||||
}
|
||||
|
||||
@@ -108,9 +120,28 @@ func (c *mxAccessGatewayClient) AcknowledgeAlarm(ctx context.Context, in *Acknow
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *mxAccessGatewayClient) StreamAlarms(ctx context.Context, in *StreamAlarmsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[AlarmFeedMessage], error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
stream, err := c.cc.NewStream(ctx, &MxAccessGateway_ServiceDesc.Streams[1], MxAccessGateway_StreamAlarms_FullMethodName, cOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
x := &grpc.GenericClientStream[StreamAlarmsRequest, AlarmFeedMessage]{ClientStream: stream}
|
||||
if err := x.ClientStream.SendMsg(in); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := x.ClientStream.CloseSend(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return x, nil
|
||||
}
|
||||
|
||||
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
|
||||
type MxAccessGateway_StreamAlarmsClient = grpc.ServerStreamingClient[AlarmFeedMessage]
|
||||
|
||||
func (c *mxAccessGatewayClient) QueryActiveAlarms(ctx context.Context, in *QueryActiveAlarmsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ActiveAlarmSnapshot], error) {
|
||||
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
|
||||
stream, err := c.cc.NewStream(ctx, &MxAccessGateway_ServiceDesc.Streams[1], MxAccessGateway_QueryActiveAlarms_FullMethodName, cOpts...)
|
||||
stream, err := c.cc.NewStream(ctx, &MxAccessGateway_ServiceDesc.Streams[2], MxAccessGateway_QueryActiveAlarms_FullMethodName, cOpts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -138,6 +169,17 @@ type MxAccessGatewayServer interface {
|
||||
Invoke(context.Context, *MxCommandRequest) (*MxCommandReply, error)
|
||||
StreamEvents(*StreamEventsRequest, grpc.ServerStreamingServer[MxEvent]) error
|
||||
AcknowledgeAlarm(context.Context, *AcknowledgeAlarmRequest) (*AcknowledgeAlarmReply, error)
|
||||
// Session-less central alarm feed. The stream opens with the current
|
||||
// active-alarm snapshot (one `active_alarm` per alarm), then a single
|
||||
// `snapshot_complete`, then a `transition` for every subsequent change.
|
||||
// Served by the gateway's always-on alarm monitor; any number of clients
|
||||
// fan out from the single monitor without opening a worker session.
|
||||
StreamAlarms(*StreamAlarmsRequest, grpc.ServerStreamingServer[AlarmFeedMessage]) error
|
||||
// Point-in-time snapshot of the currently-active alarm set served from the
|
||||
// gateway's always-on alarm monitor cache (session-less). Used after a
|
||||
// reconnect to seed Part 9 client state, or to reconcile alarms that may
|
||||
// have been missed during a transport blip. Streamed so callers can
|
||||
// begin processing without buffering the full set.
|
||||
QueryActiveAlarms(*QueryActiveAlarmsRequest, grpc.ServerStreamingServer[ActiveAlarmSnapshot]) error
|
||||
mustEmbedUnimplementedMxAccessGatewayServer()
|
||||
}
|
||||
@@ -164,6 +206,9 @@ func (UnimplementedMxAccessGatewayServer) StreamEvents(*StreamEventsRequest, grp
|
||||
func (UnimplementedMxAccessGatewayServer) AcknowledgeAlarm(context.Context, *AcknowledgeAlarmRequest) (*AcknowledgeAlarmReply, error) {
|
||||
return nil, status.Error(codes.Unimplemented, "method AcknowledgeAlarm not implemented")
|
||||
}
|
||||
func (UnimplementedMxAccessGatewayServer) StreamAlarms(*StreamAlarmsRequest, grpc.ServerStreamingServer[AlarmFeedMessage]) error {
|
||||
return status.Error(codes.Unimplemented, "method StreamAlarms not implemented")
|
||||
}
|
||||
func (UnimplementedMxAccessGatewayServer) QueryActiveAlarms(*QueryActiveAlarmsRequest, grpc.ServerStreamingServer[ActiveAlarmSnapshot]) error {
|
||||
return status.Error(codes.Unimplemented, "method QueryActiveAlarms not implemented")
|
||||
}
|
||||
@@ -271,6 +316,17 @@ func _MxAccessGateway_AcknowledgeAlarm_Handler(srv interface{}, ctx context.Cont
|
||||
return interceptor(ctx, in, info, handler)
|
||||
}
|
||||
|
||||
func _MxAccessGateway_StreamAlarms_Handler(srv interface{}, stream grpc.ServerStream) error {
|
||||
m := new(StreamAlarmsRequest)
|
||||
if err := stream.RecvMsg(m); err != nil {
|
||||
return err
|
||||
}
|
||||
return srv.(MxAccessGatewayServer).StreamAlarms(m, &grpc.GenericServerStream[StreamAlarmsRequest, AlarmFeedMessage]{ServerStream: stream})
|
||||
}
|
||||
|
||||
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
|
||||
type MxAccessGateway_StreamAlarmsServer = grpc.ServerStreamingServer[AlarmFeedMessage]
|
||||
|
||||
func _MxAccessGateway_QueryActiveAlarms_Handler(srv interface{}, stream grpc.ServerStream) error {
|
||||
m := new(QueryActiveAlarmsRequest)
|
||||
if err := stream.RecvMsg(m); err != nil {
|
||||
@@ -312,6 +368,11 @@ var MxAccessGateway_ServiceDesc = grpc.ServiceDesc{
|
||||
Handler: _MxAccessGateway_StreamEvents_Handler,
|
||||
ServerStreams: true,
|
||||
},
|
||||
{
|
||||
StreamName: "StreamAlarms",
|
||||
Handler: _MxAccessGateway_StreamAlarms_Handler,
|
||||
ServerStreams: true,
|
||||
},
|
||||
{
|
||||
StreamName: "QueryActiveAlarms",
|
||||
Handler: _MxAccessGateway_QueryActiveAlarms_Handler,
|
||||
|
||||
@@ -1179,7 +1179,7 @@ const file_mxaccess_worker_proto_rawDesc = "" +
|
||||
"\x1eWORKER_FAULT_CATEGORY_STA_HUNG\x10\t\x12(\n" +
|
||||
"$WORKER_FAULT_CATEGORY_QUEUE_OVERFLOW\x10\n" +
|
||||
"\x12*\n" +
|
||||
"&WORKER_FAULT_CATEGORY_SHUTDOWN_TIMEOUT\x10\vB\x1c\xaa\x02\x19MxGateway.Contracts.Protob\x06proto3"
|
||||
"&WORKER_FAULT_CATEGORY_SHUTDOWN_TIMEOUT\x10\vB&\xaa\x02#ZB.MOM.WW.MxGateway.Contracts.Protob\x06proto3"
|
||||
|
||||
var (
|
||||
file_mxaccess_worker_proto_rawDescOnce sync.Once
|
||||
|
||||
@@ -20,7 +20,6 @@ import (
|
||||
func TestAcknowledgeAlarmSendsRequestAndReturnsReply(t *testing.T) {
|
||||
fake := &fakeGatewayWithAlarms{
|
||||
acknowledgeReply: &pb.AcknowledgeAlarmReply{
|
||||
SessionId: "session-1",
|
||||
CorrelationId: "corr-1",
|
||||
ProtocolStatus: &pb.ProtocolStatus{
|
||||
Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
|
||||
@@ -35,7 +34,6 @@ func TestAcknowledgeAlarmSendsRequestAndReturnsReply(t *testing.T) {
|
||||
defer cleanup()
|
||||
|
||||
reply, err := client.AcknowledgeAlarm(context.Background(), &pb.AcknowledgeAlarmRequest{
|
||||
SessionId: "session-1",
|
||||
ClientCorrelationId: "corr-1",
|
||||
AlarmFullReference: "Tank01.Level.HiHi",
|
||||
Comment: "investigating",
|
||||
@@ -81,7 +79,6 @@ func TestAcknowledgeAlarmMapsUnauthenticated(t *testing.T) {
|
||||
defer cleanup()
|
||||
|
||||
_, err := client.AcknowledgeAlarm(context.Background(), &pb.AcknowledgeAlarmRequest{
|
||||
SessionId: "session-1",
|
||||
AlarmFullReference: "Tank01.Level.HiHi",
|
||||
OperatorUser: "alice",
|
||||
})
|
||||
@@ -150,8 +147,8 @@ func TestQueryActiveAlarmsPassesFilterPrefix(t *testing.T) {
|
||||
defer cleanup()
|
||||
|
||||
stream, err := client.QueryActiveAlarms(context.Background(), &pb.QueryActiveAlarmsRequest{
|
||||
SessionId: "session-1",
|
||||
AlarmFilterPrefix: "Tank01.",
|
||||
SessionId: "session-1",
|
||||
AlarmFilterPrefix: "Tank01.",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("QueryActiveAlarms() error = %v", err)
|
||||
@@ -193,7 +190,7 @@ func (s *fakeGatewayWithAlarms) AcknowledgeAlarm(ctx context.Context, req *pb.Ac
|
||||
return s.acknowledgeReply, nil
|
||||
}
|
||||
return &pb.AcknowledgeAlarmReply{
|
||||
SessionId: req.GetSessionId(),
|
||||
CorrelationId: req.GetClientCorrelationId(),
|
||||
ProtocolStatus: &pb.ProtocolStatus{
|
||||
Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK,
|
||||
},
|
||||
@@ -221,10 +218,8 @@ func newBufconnClientWithAlarms(t *testing.T, fake *fakeGatewayWithAlarms) (*Cli
|
||||
dialer := func(ctx context.Context, _ string) (net.Conn, error) {
|
||||
return listener.DialContext(ctx)
|
||||
}
|
||||
// grpc.NewClient defaults to the dns scheme; use passthrough so the
|
||||
// bufconn fake target reaches the context dialer unresolved.
|
||||
client, err := Dial(context.Background(), Options{
|
||||
Endpoint: "passthrough:///bufnet",
|
||||
Endpoint: "bufnet",
|
||||
APIKey: "test-api-key",
|
||||
Plaintext: true,
|
||||
DialOptions: []grpc.DialOption{grpc.WithContextDialer(dialer)},
|
||||
|
||||
@@ -19,7 +19,6 @@ import (
|
||||
|
||||
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/connectivity"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/protobuf/types/known/durationpb"
|
||||
@@ -37,36 +36,22 @@ type Client struct {
|
||||
opts Options
|
||||
}
|
||||
|
||||
// Dial opens a gRPC connection to the gateway and configures auth metadata
|
||||
// and transport security.
|
||||
//
|
||||
// The connection is created lazily with grpc.NewClient: the channel is not
|
||||
// established until the first RPC (or the readiness probe below) needs it, so
|
||||
// a gateway that is briefly unavailable at Dial time no longer turns into a
|
||||
// hard error — the connection recovers when the gateway comes up. To preserve
|
||||
// fail-fast behavior, Dial then runs an explicit readiness probe bounded by
|
||||
// DialTimeout (default 10s, or ctx's deadline when sooner): it triggers the
|
||||
// initial connect and waits for the channel to reach Ready, returning a
|
||||
// *GatewayError if the gateway cannot be reached in that window. Cancelling
|
||||
// ctx aborts the probe.
|
||||
// Dial opens a gRPC connection to the gateway and configures auth metadata,
|
||||
// transport security, and blocking dial cancellation from ctx.
|
||||
func Dial(ctx context.Context, opts Options) (*Client, error) {
|
||||
conn, err := dial(ctx, opts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return NewClient(conn, opts), nil
|
||||
}
|
||||
|
||||
// dial builds the shared gRPC connection used by both Client and GalaxyClient:
|
||||
// it resolves transport credentials, assembles dial options, creates a lazy
|
||||
// connection with grpc.NewClient, and runs the DialTimeout-bounded readiness
|
||||
// probe so callers still fail fast when the gateway is unreachable.
|
||||
func dial(ctx context.Context, opts Options) (*grpc.ClientConn, error) {
|
||||
if opts.Endpoint == "" {
|
||||
return nil, errors.New("mxgateway: endpoint is required")
|
||||
}
|
||||
|
||||
dialCtx := ctx
|
||||
cancel := func() {}
|
||||
if opts.DialTimeout > 0 {
|
||||
dialCtx, cancel = context.WithTimeout(ctx, opts.DialTimeout)
|
||||
} else if _, ok := ctx.Deadline(); !ok {
|
||||
dialCtx, cancel = context.WithTimeout(ctx, defaultDialTimeout)
|
||||
}
|
||||
defer cancel()
|
||||
|
||||
transportCredentials, err := resolveTransportCredentials(opts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -76,46 +61,16 @@ func dial(ctx context.Context, opts Options) (*grpc.ClientConn, error) {
|
||||
grpc.WithTransportCredentials(transportCredentials),
|
||||
grpc.WithUnaryInterceptor(unaryAuthInterceptor(opts.APIKey)),
|
||||
grpc.WithStreamInterceptor(streamAuthInterceptor(opts.APIKey)),
|
||||
grpc.WithBlock(),
|
||||
}
|
||||
dialOptions = append(dialOptions, opts.DialOptions...)
|
||||
|
||||
conn, err := grpc.NewClient(opts.Endpoint, dialOptions...)
|
||||
conn, err := grpc.DialContext(dialCtx, opts.Endpoint, dialOptions...)
|
||||
if err != nil {
|
||||
return nil, &GatewayError{Op: "dial", Err: err}
|
||||
}
|
||||
|
||||
if err := waitForReady(ctx, conn, opts.DialTimeout); err != nil {
|
||||
_ = conn.Close()
|
||||
return nil, &GatewayError{Op: "dial", Err: err}
|
||||
}
|
||||
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
// waitForReady triggers the initial connect on conn and blocks until the
|
||||
// channel reaches connectivity.Ready, the timeout elapses, or ctx is
|
||||
// cancelled. The wait is bounded by dialTimeout when positive, otherwise by
|
||||
// ctx's existing deadline, otherwise by defaultDialTimeout.
|
||||
func waitForReady(ctx context.Context, conn *grpc.ClientConn, dialTimeout time.Duration) error {
|
||||
probeCtx := ctx
|
||||
cancel := func() {}
|
||||
if dialTimeout > 0 {
|
||||
probeCtx, cancel = context.WithTimeout(ctx, dialTimeout)
|
||||
} else if _, ok := ctx.Deadline(); !ok {
|
||||
probeCtx, cancel = context.WithTimeout(ctx, defaultDialTimeout)
|
||||
}
|
||||
defer cancel()
|
||||
|
||||
conn.Connect()
|
||||
for {
|
||||
state := conn.GetState()
|
||||
if state == connectivity.Ready {
|
||||
return nil
|
||||
}
|
||||
if !conn.WaitForStateChange(probeCtx, state) {
|
||||
return probeCtx.Err()
|
||||
}
|
||||
}
|
||||
return NewClient(conn, opts), nil
|
||||
}
|
||||
|
||||
// NewClient wraps an existing gRPC connection. The caller owns closing conn
|
||||
@@ -233,15 +188,7 @@ func (c *Client) Close() error {
|
||||
}
|
||||
|
||||
func (c *Client) callContext(ctx context.Context) (context.Context, context.CancelFunc) {
|
||||
return callContext(ctx, c.opts.CallTimeout)
|
||||
}
|
||||
|
||||
// callContext derives a per-RPC context from ctx, applying callTimeout: zero
|
||||
// uses defaultCallTimeout, a negative value disables the bound entirely, and a
|
||||
// caller-supplied deadline that is already sooner than the derived timeout is
|
||||
// kept as-is rather than being lengthened.
|
||||
func callContext(ctx context.Context, callTimeout time.Duration) (context.Context, context.CancelFunc) {
|
||||
timeout := callTimeout
|
||||
timeout := c.opts.CallTimeout
|
||||
if timeout == 0 {
|
||||
timeout = defaultCallTimeout
|
||||
}
|
||||
|
||||
@@ -117,7 +117,7 @@ func TestEventsAfterCancelsStreamWhenCompatibilityChannelIsAbandoned(t *testing.
|
||||
fake := &fakeGatewayServer{
|
||||
streamStarted: make(chan struct{}),
|
||||
streamDone: make(chan struct{}),
|
||||
streamEventCount: 256,
|
||||
streamEventCount: 64,
|
||||
}
|
||||
client, cleanup := newBufconnClient(t, fake)
|
||||
defer cleanup()
|
||||
@@ -135,25 +135,12 @@ func TestEventsAfterCancelsStreamWhenCompatibilityChannelIsAbandoned(t *testing.
|
||||
t.Fatal("compatibility event stream did not stop after result channel filled")
|
||||
}
|
||||
|
||||
// A slow consumer that abandons the buffer must still receive an explicit
|
||||
// terminal overflow error before the channel closes, so it can tell
|
||||
// "events dropped" apart from "stream ended normally".
|
||||
var sawOverflow bool
|
||||
for {
|
||||
select {
|
||||
case result, ok := <-events:
|
||||
case _, ok := <-events:
|
||||
if !ok {
|
||||
if !sawOverflow {
|
||||
t.Fatal("compatibility event channel closed without an ErrEventBufferOverflow result")
|
||||
}
|
||||
return
|
||||
}
|
||||
if result.Err != nil {
|
||||
if !errors.Is(result.Err, ErrEventBufferOverflow) {
|
||||
t.Fatalf("terminal result error = %v, want ErrEventBufferOverflow", result.Err)
|
||||
}
|
||||
sawOverflow = true
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("compatibility event channel did not close")
|
||||
}
|
||||
@@ -292,11 +279,8 @@ func newBufconnClient(t *testing.T, fake *fakeGatewayServer) (*Client, func()) {
|
||||
dialer := func(ctx context.Context, _ string) (net.Conn, error) {
|
||||
return listener.DialContext(ctx)
|
||||
}
|
||||
// grpc.NewClient defaults the target scheme to dns; the bufconn fake name
|
||||
// is not DNS-resolvable, so use the passthrough scheme to hand the target
|
||||
// straight to the context dialer.
|
||||
client, err := Dial(context.Background(), Options{
|
||||
Endpoint: "passthrough:///bufnet",
|
||||
Endpoint: "bufnet",
|
||||
APIKey: "test-api-key",
|
||||
Plaintext: true,
|
||||
DialOptions: []grpc.DialOption{
|
||||
|
||||
@@ -1,401 +0,0 @@
|
||||
package mxgateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"net"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/types/known/timestamppb"
|
||||
)
|
||||
|
||||
// --- Client.Go-008: resolveTransportCredentials precedence -----------------
|
||||
|
||||
// TestResolveTransportCredentialsPrecedence covers every branch of
|
||||
// resolveTransportCredentials, which previously only had the Plaintext path
|
||||
// exercised.
|
||||
func TestResolveTransportCredentialsPrecedence(t *testing.T) {
|
||||
custom := insecure.NewCredentials()
|
||||
|
||||
t.Run("TransportCredentialsWins", func(t *testing.T) {
|
||||
creds, err := resolveTransportCredentials(Options{
|
||||
TransportCredentials: custom,
|
||||
Plaintext: true, // must be ignored
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if creds != custom {
|
||||
t.Fatal("expected the explicit TransportCredentials to be returned as-is")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("Plaintext", func(t *testing.T) {
|
||||
creds, err := resolveTransportCredentials(Options{Plaintext: true})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if got := creds.Info().SecurityProtocol; got != "insecure" {
|
||||
t.Fatalf("expected insecure credentials, got security protocol %q", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("CACertFileMissingErrors", func(t *testing.T) {
|
||||
_, err := resolveTransportCredentials(Options{CACertFile: "does-not-exist.pem"})
|
||||
if err == nil {
|
||||
t.Fatal("expected an error for a missing CA cert file")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("TLSConfigWithServerNameOverride", func(t *testing.T) {
|
||||
creds, err := resolveTransportCredentials(Options{
|
||||
TLSConfig: &tls.Config{MinVersion: tls.VersionTLS13},
|
||||
ServerNameOverride: "gateway.internal",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if got := creds.Info().ServerName; got != "gateway.internal" {
|
||||
t.Fatalf("expected ServerName override to be applied, got %q", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("DefaultTLSFloor", func(t *testing.T) {
|
||||
creds, err := resolveTransportCredentials(Options{ServerNameOverride: "host"})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if got := creds.Info().SecurityProtocol; got != "tls" {
|
||||
t.Fatalf("expected the default TLS credentials, got %q", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// TestResolveTransportCredentialsDoesNotMutateTLSConfig confirms the supplied
|
||||
// TLSConfig is cloned, not mutated, when ServerNameOverride is applied.
|
||||
func TestResolveTransportCredentialsDoesNotMutateTLSConfig(t *testing.T) {
|
||||
cfg := &tls.Config{MinVersion: tls.VersionTLS12}
|
||||
if _, err := resolveTransportCredentials(Options{
|
||||
TLSConfig: cfg,
|
||||
ServerNameOverride: "override",
|
||||
}); err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if cfg.ServerName != "" {
|
||||
t.Fatalf("resolveTransportCredentials mutated the caller's TLSConfig (ServerName=%q)", cfg.ServerName)
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Go-008: callContext deadline arithmetic ------------------------
|
||||
|
||||
// TestCallContextDeadlineArithmetic covers the shared callContext deadline
|
||||
// logic, including the negative-timeout disable case and the
|
||||
// caller-deadline-is-sooner case.
|
||||
func TestCallContextDeadlineArithmetic(t *testing.T) {
|
||||
t.Run("ZeroUsesDefault", func(t *testing.T) {
|
||||
ctx, cancel := callContext(context.Background(), 0)
|
||||
defer cancel()
|
||||
deadline, ok := ctx.Deadline()
|
||||
if !ok {
|
||||
t.Fatal("expected a deadline for the default timeout")
|
||||
}
|
||||
remaining := time.Until(deadline)
|
||||
if remaining <= 0 || remaining > defaultCallTimeout+time.Second {
|
||||
t.Fatalf("default deadline out of range: %v", remaining)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("NegativeDisablesBound", func(t *testing.T) {
|
||||
base := context.Background()
|
||||
ctx, cancel := callContext(base, -1)
|
||||
defer cancel()
|
||||
if _, ok := ctx.Deadline(); ok {
|
||||
t.Fatal("a negative timeout must disable the deadline entirely")
|
||||
}
|
||||
if ctx != base {
|
||||
t.Fatal("a negative timeout must return the caller context unchanged")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("PositiveAppliesTimeout", func(t *testing.T) {
|
||||
ctx, cancel := callContext(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
deadline, ok := ctx.Deadline()
|
||||
if !ok {
|
||||
t.Fatal("expected a deadline")
|
||||
}
|
||||
remaining := time.Until(deadline)
|
||||
if remaining <= 0 || remaining > 5*time.Second+time.Second {
|
||||
t.Fatalf("deadline out of range: %v", remaining)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("CallerDeadlineSoonerIsKept", func(t *testing.T) {
|
||||
base, baseCancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||
defer baseCancel()
|
||||
ctx, cancel := callContext(base, 30*time.Second)
|
||||
defer cancel()
|
||||
if ctx != base {
|
||||
t.Fatal("a caller deadline sooner than the timeout must be kept as-is")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("CallerDeadlineLaterIsShortened", func(t *testing.T) {
|
||||
base, baseCancel := context.WithTimeout(context.Background(), time.Hour)
|
||||
defer baseCancel()
|
||||
ctx, cancel := callContext(base, time.Second)
|
||||
defer cancel()
|
||||
deadline, ok := ctx.Deadline()
|
||||
if !ok {
|
||||
t.Fatal("expected a deadline")
|
||||
}
|
||||
if remaining := time.Until(deadline); remaining > 2*time.Second {
|
||||
t.Fatalf("expected the shorter timeout to win, got %v remaining", remaining)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// --- Client.Go-008: NativeValue / NativeArray edge branches ----------------
|
||||
|
||||
// TestNativeValueEdgeKinds covers the array, raw-bytes, null, and
|
||||
// nil-input branches of NativeValue.
|
||||
func TestNativeValueEdgeKinds(t *testing.T) {
|
||||
t.Run("NilInput", func(t *testing.T) {
|
||||
got, err := NativeValue(nil)
|
||||
if err != nil || got != nil {
|
||||
t.Fatalf("NativeValue(nil) = (%v, %v), want (nil, nil)", got, err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("ExplicitNull", func(t *testing.T) {
|
||||
got, err := NativeValue(&pb.MxValue{IsNull: true})
|
||||
if err != nil || got != nil {
|
||||
t.Fatalf("NativeValue(null) = (%v, %v), want (nil, nil)", got, err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("RawBytes", func(t *testing.T) {
|
||||
raw := []byte{0x01, 0x02, 0x03}
|
||||
got, err := NativeValue(&pb.MxValue{Kind: &pb.MxValue_RawValue{RawValue: raw}})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
gotBytes, ok := got.([]byte)
|
||||
if !ok || !reflect.DeepEqual(gotBytes, raw) {
|
||||
t.Fatalf("NativeValue raw = %v, want %v", got, raw)
|
||||
}
|
||||
// The result must be a copy, not aliasing the protobuf field.
|
||||
gotBytes[0] = 0xFF
|
||||
if raw[0] != 0x01 {
|
||||
t.Fatal("NativeValue raw result aliases the protobuf backing array")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("ArrayValue", func(t *testing.T) {
|
||||
value := &pb.MxValue{Kind: &pb.MxValue_ArrayValue{
|
||||
ArrayValue: &pb.MxArray{Values: &pb.MxArray_Int32Values{
|
||||
Int32Values: &pb.Int32Array{Values: []int32{7, 8}},
|
||||
}},
|
||||
}}
|
||||
got, err := NativeValue(value)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if !reflect.DeepEqual(got, []int32{7, 8}) {
|
||||
t.Fatalf("NativeValue array = %v, want [7 8]", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// TestNativeArrayEdgeKinds covers the nil, raw-bytes, timestamp-with-nil, and
|
||||
// unsupported-kind branches of NativeArray.
|
||||
func TestNativeArrayEdgeKinds(t *testing.T) {
|
||||
t.Run("NilInput", func(t *testing.T) {
|
||||
got, err := NativeArray(nil)
|
||||
if err != nil || got != nil {
|
||||
t.Fatalf("NativeArray(nil) = (%v, %v), want (nil, nil)", got, err)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("RawValues", func(t *testing.T) {
|
||||
got, err := NativeArray(&pb.MxArray{Values: &pb.MxArray_RawValues{
|
||||
RawValues: &pb.RawArray{Values: [][]byte{{0x0A}, {0x0B}}},
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
want := [][]byte{{0x0A}, {0x0B}}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("NativeArray raw = %v, want %v", got, want)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("TimestampWithNilEntry", func(t *testing.T) {
|
||||
got, err := NativeArray(&pb.MxArray{Values: &pb.MxArray_TimestampValues{
|
||||
TimestampValues: &pb.TimestampArray{Values: []*timestamppb.Timestamp{nil}},
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
times, ok := got.([]time.Time)
|
||||
if !ok || len(times) != 1 || !times[0].IsZero() {
|
||||
t.Fatalf("NativeArray timestamp-with-nil = %v, want [zero-time]", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("UnsupportedKind", func(t *testing.T) {
|
||||
// An MxArray with no oneof set hits the default branch.
|
||||
_, err := NativeArray(&pb.MxArray{})
|
||||
if err == nil {
|
||||
t.Fatal("expected an error for an MxArray with no values set")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "unsupported array value kind") {
|
||||
t.Fatalf("unexpected error text: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// TestNativeValueUnsupportedKind covers the default branch of NativeValue.
|
||||
func TestNativeValueUnsupportedKind(t *testing.T) {
|
||||
// An MxValue with no oneof Kind set and IsNull false hits the default.
|
||||
_, err := NativeValue(&pb.MxValue{})
|
||||
if err == nil {
|
||||
t.Fatal("expected an error for an MxValue with no kind set")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "unsupported value kind") {
|
||||
t.Fatalf("unexpected error text: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Go-005: dial migration -----------------------------------------
|
||||
|
||||
// TestDialFailsFastWhenGatewayUnreachable confirms that after the migration to
|
||||
// grpc.NewClient the DialTimeout-bounded readiness probe still fails fast (and
|
||||
// wraps the failure in *GatewayError) when the gateway cannot be reached.
|
||||
func TestDialFailsFastWhenGatewayUnreachable(t *testing.T) {
|
||||
dialer := func(ctx context.Context, _ string) (net.Conn, error) {
|
||||
return nil, errors.New("connection refused")
|
||||
}
|
||||
start := time.Now()
|
||||
client, err := Dial(context.Background(), Options{
|
||||
Endpoint: "passthrough:///unreachable",
|
||||
APIKey: "k",
|
||||
Plaintext: true,
|
||||
DialTimeout: 500 * time.Millisecond,
|
||||
DialOptions: []grpc.DialOption{grpc.WithContextDialer(dialer)},
|
||||
})
|
||||
elapsed := time.Since(start)
|
||||
if err == nil {
|
||||
client.Close()
|
||||
t.Fatal("expected Dial to fail for an unreachable gateway")
|
||||
}
|
||||
var gwErr *GatewayError
|
||||
if !errors.As(err, &gwErr) || gwErr.Op != "dial" {
|
||||
t.Fatalf("expected a *GatewayError with Op=dial, got %#v", err)
|
||||
}
|
||||
if elapsed > 5*time.Second {
|
||||
t.Fatalf("Dial did not honor DialTimeout: took %v", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDialReadinessProbeReachesReady confirms the readiness probe succeeds
|
||||
// against a live (bufconn) gateway, i.e. the lazy grpc.NewClient connection is
|
||||
// driven to Ready before Dial returns.
|
||||
func TestDialReadinessProbeReachesReady(t *testing.T) {
|
||||
client, cleanup := newBufconnClient(t, &fakeGatewayServer{
|
||||
openReply: &pb.OpenSessionReply{},
|
||||
})
|
||||
defer cleanup()
|
||||
if client == nil {
|
||||
t.Fatal("expected a connected client")
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Go-006: error taxonomy ----------------------------------------
|
||||
|
||||
// TestGatewayErrorCode confirms GatewayError.Code surfaces the wrapped gRPC
|
||||
// status code without the caller unwrapping it.
|
||||
func TestGatewayErrorCode(t *testing.T) {
|
||||
var nilErr *GatewayError
|
||||
if got := nilErr.Code(); got != codes.OK {
|
||||
t.Fatalf("nil GatewayError.Code() = %v, want OK", got)
|
||||
}
|
||||
|
||||
gwErr := &GatewayError{Op: "invoke", Err: status.Error(codes.Unavailable, "down")}
|
||||
if got := gwErr.Code(); got != codes.Unavailable {
|
||||
t.Fatalf("GatewayError.Code() = %v, want Unavailable", got)
|
||||
}
|
||||
|
||||
plain := &GatewayError{Op: "dial", Err: errors.New("boom")}
|
||||
if got := plain.Code(); got != codes.Unknown {
|
||||
t.Fatalf("GatewayError.Code() for a non-status error = %v, want Unknown", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestIsTransient verifies the transient/permanent classification including
|
||||
// the unwrap-through-GatewayError path.
|
||||
func TestIsTransient(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{name: "nil", err: nil, want: false},
|
||||
{name: "unavailable wrapped", err: &GatewayError{Op: "invoke", Err: status.Error(codes.Unavailable, "x")}, want: true},
|
||||
{name: "deadline wrapped", err: &GatewayError{Op: "invoke", Err: status.Error(codes.DeadlineExceeded, "x")}, want: true},
|
||||
{name: "resource exhausted", err: &GatewayError{Err: status.Error(codes.ResourceExhausted, "x")}, want: true},
|
||||
{name: "unauthenticated permanent", err: &GatewayError{Err: status.Error(codes.Unauthenticated, "x")}, want: false},
|
||||
{name: "invalid argument permanent", err: &GatewayError{Err: status.Error(codes.InvalidArgument, "x")}, want: false},
|
||||
{name: "bare status unavailable", err: status.Error(codes.Unavailable, "x"), want: true},
|
||||
{name: "plain error", err: errors.New("nope"), want: false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := IsTransient(tt.err); got != tt.want {
|
||||
t.Fatalf("IsTransient(%v) = %v, want %v", tt.err, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Go-007: correlation id fallback --------------------------------
|
||||
|
||||
// TestNewCorrelationIDUsesRandEntropy confirms the happy path yields a
|
||||
// 32-hex-character id.
|
||||
func TestNewCorrelationIDUsesRandEntropy(t *testing.T) {
|
||||
id := newCorrelationID()
|
||||
if len(id) != 32 {
|
||||
t.Fatalf("expected a 32-char hex id, got %q (len %d)", id, len(id))
|
||||
}
|
||||
}
|
||||
|
||||
// TestNewCorrelationIDFallsBackOnRandFailure reproduces Client.Go-007: when
|
||||
// crypto/rand fails, newCorrelationID must not return an empty string but a
|
||||
// unique, non-empty fallback id so the command stays traceable.
|
||||
func TestNewCorrelationIDFallsBackOnRandFailure(t *testing.T) {
|
||||
original := randRead
|
||||
randRead = func([]byte) (int, error) { return 0, errors.New("entropy unavailable") }
|
||||
defer func() { randRead = original }()
|
||||
|
||||
first := newCorrelationID()
|
||||
second := newCorrelationID()
|
||||
|
||||
if first == "" || second == "" {
|
||||
t.Fatal("newCorrelationID returned an empty id on rand failure")
|
||||
}
|
||||
if !strings.HasPrefix(first, "fallback-") {
|
||||
t.Fatalf("expected a fallback- prefixed id, got %q", first)
|
||||
}
|
||||
if first == second {
|
||||
t.Fatalf("fallback correlation ids must be unique, got %q twice", first)
|
||||
}
|
||||
}
|
||||
@@ -1,22 +1,11 @@
|
||||
package mxgateway
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
// ErrEventBufferOverflow is the terminal error delivered on the compatibility
|
||||
// event channel returned by Session.Events / Session.EventsAfter when a slow
|
||||
// consumer lets the bounded result buffer fill. It signals that the stream was
|
||||
// cancelled and events were dropped, so a consumer can tell an overflow apart
|
||||
// from a normal end-of-stream. Use Session.SubscribeEvents to block instead of
|
||||
// dropping.
|
||||
var ErrEventBufferOverflow = errors.New("mxgateway: event buffer overflow; compatibility stream cancelled and events dropped")
|
||||
|
||||
// GatewayError wraps transport-level gRPC failures.
|
||||
type GatewayError struct {
|
||||
// Op names the operation that failed (for example "dial" or "invoke").
|
||||
@@ -44,45 +33,6 @@ func (e *GatewayError) Unwrap() error {
|
||||
return e.Err
|
||||
}
|
||||
|
||||
// Code returns the gRPC status code of the wrapped transport error. It returns
|
||||
// codes.OK when the error is nil and codes.Unknown when the wrapped error does
|
||||
// not carry a gRPC status. Callers can use it to write retry, timeout, and
|
||||
// auth handling without manually unwrapping and re-parsing the error.
|
||||
func (e *GatewayError) Code() codes.Code {
|
||||
if e == nil || e.Err == nil {
|
||||
return codes.OK
|
||||
}
|
||||
return status.Code(e.Err)
|
||||
}
|
||||
|
||||
// IsTransient reports whether err is a transport failure that may succeed on
|
||||
// retry — for example a gateway that is briefly Unavailable or a call that
|
||||
// hit a DeadlineExceeded. Permanent failures (Unauthenticated, PermissionDenied,
|
||||
// InvalidArgument, NotFound, and similar) return false. It unwraps through
|
||||
// *GatewayError and any other error chain carrying a gRPC status, so callers
|
||||
// do not need to call status.Code themselves.
|
||||
func IsTransient(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
switch transientCode(err) {
|
||||
case codes.Unavailable, codes.DeadlineExceeded, codes.ResourceExhausted, codes.Aborted:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// transientCode extracts a gRPC status code from err, preferring a wrapped
|
||||
// *GatewayError's Code and otherwise falling back to status.Code on the chain.
|
||||
func transientCode(err error) codes.Code {
|
||||
var gatewayErr *GatewayError
|
||||
if errors.As(err, &gatewayErr) {
|
||||
return gatewayErr.Code()
|
||||
}
|
||||
return status.Code(err)
|
||||
}
|
||||
|
||||
// CommandError reports a non-OK gateway protocol status and keeps the raw
|
||||
// command reply when one exists.
|
||||
type CommandError struct {
|
||||
@@ -135,12 +85,8 @@ func (e *MxAccessError) Error() string {
|
||||
}
|
||||
|
||||
// Unwrap returns the wrapped CommandError, when one is present.
|
||||
//
|
||||
// When Command is nil (the HRESULT / MxStatusProxy path) it returns an
|
||||
// untyped nil rather than a typed-nil *CommandError, so errors.As does not
|
||||
// bind a nil pointer that a caller would then panic on.
|
||||
func (e *MxAccessError) Unwrap() error {
|
||||
if e == nil || e.Command == nil {
|
||||
if e == nil {
|
||||
return nil
|
||||
}
|
||||
return e.Command
|
||||
|
||||
@@ -1,42 +0,0 @@
|
||||
package mxgateway
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestMxAccessErrorUnwrapHResultPathNoTypedNilCommandError reproduces
|
||||
// Client.Go-001: an MxAccessError built via the HRESULT / MxStatusProxy path
|
||||
// leaves Command nil. Unwrap must not hand back a typed-nil *CommandError,
|
||||
// because errors.As would then succeed while binding a nil pointer and a
|
||||
// caller dereferencing it would panic.
|
||||
func TestMxAccessErrorUnwrapHResultPathNoTypedNilCommandError(t *testing.T) {
|
||||
hresult := int32(-2147467259) // 0x80004005, a failing HRESULT.
|
||||
reply := &MxCommandReply{Hresult: &hresult}
|
||||
|
||||
err := EnsureMxAccessSuccess("invoke", reply)
|
||||
if err == nil {
|
||||
t.Fatal("expected MxAccessError for a failing HRESULT, got nil")
|
||||
}
|
||||
|
||||
var ce *CommandError
|
||||
if errors.As(err, &ce) {
|
||||
t.Fatalf("errors.As bound *CommandError from an HRESULT-only MxAccessError (ce=%v); "+
|
||||
"a caller dereferencing ce.Status would panic", ce)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMxAccessErrorUnwrapPopulatedCommand confirms the non-nil Command path
|
||||
// still unwraps to the wrapped *CommandError.
|
||||
func TestMxAccessErrorUnwrapPopulatedCommand(t *testing.T) {
|
||||
command := &CommandError{Op: "invoke"}
|
||||
err := &MxAccessError{Command: command}
|
||||
|
||||
var ce *CommandError
|
||||
if !errors.As(err, &ce) {
|
||||
t.Fatal("errors.As failed to bind the populated *CommandError")
|
||||
}
|
||||
if ce != command {
|
||||
t.Fatalf("errors.As bound an unexpected *CommandError: got %v want %v", ce, command)
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@ package mxgateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"time"
|
||||
|
||||
@@ -55,13 +56,39 @@ type GalaxyClient struct {
|
||||
|
||||
// DialGalaxy opens a gRPC connection to the gateway for the Galaxy Repository
|
||||
// service. It applies the same authentication metadata, transport security,
|
||||
// lazy connection, and DialTimeout-bounded readiness probe as Dial.
|
||||
// and dial-timeout behavior as Dial.
|
||||
func DialGalaxy(ctx context.Context, opts Options) (*GalaxyClient, error) {
|
||||
conn, err := dial(ctx, opts)
|
||||
if opts.Endpoint == "" {
|
||||
return nil, errors.New("mxgateway: endpoint is required")
|
||||
}
|
||||
|
||||
dialCtx := ctx
|
||||
cancel := func() {}
|
||||
if opts.DialTimeout > 0 {
|
||||
dialCtx, cancel = context.WithTimeout(ctx, opts.DialTimeout)
|
||||
} else if _, ok := ctx.Deadline(); !ok {
|
||||
dialCtx, cancel = context.WithTimeout(ctx, defaultDialTimeout)
|
||||
}
|
||||
defer cancel()
|
||||
|
||||
transportCredentials, err := resolveTransportCredentials(opts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
dialOptions := []grpc.DialOption{
|
||||
grpc.WithTransportCredentials(transportCredentials),
|
||||
grpc.WithUnaryInterceptor(unaryAuthInterceptor(opts.APIKey)),
|
||||
grpc.WithStreamInterceptor(streamAuthInterceptor(opts.APIKey)),
|
||||
grpc.WithBlock(),
|
||||
}
|
||||
dialOptions = append(dialOptions, opts.DialOptions...)
|
||||
|
||||
conn, err := grpc.DialContext(dialCtx, opts.Endpoint, dialOptions...)
|
||||
if err != nil {
|
||||
return nil, &GatewayError{Op: "dial", Err: err}
|
||||
}
|
||||
|
||||
return NewGalaxyClient(conn, opts), nil
|
||||
}
|
||||
|
||||
@@ -212,5 +239,18 @@ func (c *GalaxyClient) Close() error {
|
||||
}
|
||||
|
||||
func (c *GalaxyClient) callContext(ctx context.Context) (context.Context, context.CancelFunc) {
|
||||
return callContext(ctx, c.opts.CallTimeout)
|
||||
timeout := c.opts.CallTimeout
|
||||
if timeout == 0 {
|
||||
timeout = defaultCallTimeout
|
||||
}
|
||||
if timeout < 0 {
|
||||
return ctx, func() {}
|
||||
}
|
||||
if deadline, ok := ctx.Deadline(); ok {
|
||||
timeoutDeadline := time.Now().Add(timeout)
|
||||
if deadline.Before(timeoutDeadline) {
|
||||
return ctx, func() {}
|
||||
}
|
||||
}
|
||||
return context.WithTimeout(ctx, timeout)
|
||||
}
|
||||
|
||||
@@ -55,8 +55,8 @@ func TestGalaxyGetLastDeployTimeReturnsTimestampWhenPresent(t *testing.T) {
|
||||
want := time.Date(2026, 4, 28, 12, 34, 56, 0, time.UTC)
|
||||
fake := &fakeGalaxyServer{
|
||||
deployReply: &pb.GetLastDeployTimeReply{
|
||||
Present: true,
|
||||
TimeOfLastDeploy: timestamppb.New(want),
|
||||
Present: true,
|
||||
TimeOfLastDeploy: timestamppb.New(want),
|
||||
},
|
||||
}
|
||||
client, cleanup := newGalaxyBufconnClient(t, fake)
|
||||
@@ -348,10 +348,8 @@ func newGalaxyBufconnClient(t *testing.T, fake *fakeGalaxyServer) (*GalaxyClient
|
||||
dialer := func(ctx context.Context, _ string) (net.Conn, error) {
|
||||
return listener.DialContext(ctx)
|
||||
}
|
||||
// grpc.NewClient defaults to the dns scheme; use passthrough so the
|
||||
// bufconn fake target reaches the context dialer unresolved.
|
||||
client, err := DialGalaxy(context.Background(), Options{
|
||||
Endpoint: "passthrough:///bufnet",
|
||||
Endpoint: "bufnet",
|
||||
APIKey: "test-api-key",
|
||||
Plaintext: true,
|
||||
DialOptions: []grpc.DialOption{
|
||||
|
||||
@@ -8,8 +8,6 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
|
||||
"google.golang.org/grpc/codes"
|
||||
@@ -492,7 +490,7 @@ func ensureBulkSize(name string, length int) error {
|
||||
|
||||
func sendEventResult(
|
||||
ctx context.Context,
|
||||
results chan EventResult,
|
||||
results chan<- EventResult,
|
||||
result EventResult,
|
||||
cancelWhenBufferFull bool,
|
||||
cancel context.CancelFunc,
|
||||
@@ -504,12 +502,7 @@ func sendEventResult(
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
default:
|
||||
// The bounded compatibility buffer is full. Cancel the stream and
|
||||
// deliver an explicit terminal overflow error so a slow consumer
|
||||
// can tell dropped events apart from a normal end-of-stream,
|
||||
// rather than seeing the channel close silently.
|
||||
cancel()
|
||||
deliverTerminalResult(results, EventResult{Err: ErrEventBufferOverflow})
|
||||
return false
|
||||
}
|
||||
}
|
||||
@@ -522,25 +515,6 @@ func sendEventResult(
|
||||
}
|
||||
}
|
||||
|
||||
// deliverTerminalResult places result on a full buffered channel by evicting
|
||||
// one of the oldest buffered events to make room. The caller closes results
|
||||
// afterwards, so the terminal result becomes the consumer's last item.
|
||||
func deliverTerminalResult(results chan EventResult, result EventResult) {
|
||||
for {
|
||||
select {
|
||||
case results <- result:
|
||||
return
|
||||
default:
|
||||
}
|
||||
select {
|
||||
case <-results:
|
||||
default:
|
||||
// Another receiver drained the channel between the send and
|
||||
// receive attempts; retry the send.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Session) invokeCommand(ctx context.Context, command *MxCommand) (*MxCommandReply, error) {
|
||||
return s.client.Invoke(ctx, &pb.MxCommandRequest{
|
||||
SessionId: s.ID(),
|
||||
@@ -549,25 +523,10 @@ func (s *Session) invokeCommand(ctx context.Context, command *MxCommand) (*MxCom
|
||||
})
|
||||
}
|
||||
|
||||
// correlationIDCounter backs the deterministic fallback id used when
|
||||
// crypto/rand is unavailable, so every command still carries a unique,
|
||||
// traceable correlation id.
|
||||
var correlationIDCounter atomic.Uint64
|
||||
|
||||
// randRead is the entropy source for newCorrelationID. It is a package
|
||||
// variable solely so tests can simulate a crypto/rand failure.
|
||||
var randRead = rand.Read
|
||||
|
||||
// newCorrelationID returns a unique correlation id for an MxCommandRequest.
|
||||
// It prefers 16 bytes of crypto/rand entropy; if rand.Read fails (rare) it
|
||||
// falls back to a "fallback-" prefixed id built from the current time and a
|
||||
// process-wide monotonic counter rather than returning an empty string, which
|
||||
// would leave the command untraceable in gateway logs.
|
||||
func newCorrelationID() string {
|
||||
var buffer [16]byte
|
||||
if _, err := randRead(buffer[:]); err != nil {
|
||||
return fmt.Sprintf("fallback-%x-%x",
|
||||
time.Now().UnixNano(), correlationIDCounter.Add(1))
|
||||
if _, err := rand.Read(buffer[:]); err != nil {
|
||||
return ""
|
||||
}
|
||||
return hex.EncodeToString(buffer[:])
|
||||
}
|
||||
|
||||
@@ -18,11 +18,11 @@ clients/java/
|
||||
settings.gradle
|
||||
build.gradle
|
||||
src/main/generated/
|
||||
mxgateway-client/
|
||||
zb-mom-ww-mxgateway-client/
|
||||
build.gradle
|
||||
src/main/java/com/dohertylan/mxgateway/client/
|
||||
src/test/java/com/dohertylan/mxgateway/client/
|
||||
mxgateway-cli/
|
||||
zb-mom-ww-mxgateway-cli/
|
||||
build.gradle
|
||||
src/main/java/com/dohertylan/mxgateway/cli/
|
||||
```
|
||||
|
||||
+6
-33
@@ -10,12 +10,12 @@ clients/java/
|
||||
settings.gradle
|
||||
build.gradle
|
||||
src/main/generated/
|
||||
mxgateway-client/
|
||||
mxgateway-cli/
|
||||
zb-mom-ww-mxgateway-client/
|
||||
zb-mom-ww-mxgateway-cli/
|
||||
```
|
||||
|
||||
`mxgateway-client` generates Java protobuf and gRPC sources from
|
||||
`../../src/MxGateway.Contracts/Protos`. The Gradle protobuf plugin writes those
|
||||
`../../src/ZB.MOM.WW.MxGateway.Contracts/Protos`. The Gradle protobuf plugin writes those
|
||||
generated sources under `src/main/generated`, which matches the client proto
|
||||
manifest in `../proto/proto-inputs.json`. Do not edit generated files by hand.
|
||||
|
||||
@@ -62,37 +62,10 @@ underlying protobuf messages. `MxGatewayCommandException` and
|
||||
`MxAccessException` preserve the raw `MxCommandReply` when the gateway returns a
|
||||
data-bearing MXAccess failure.
|
||||
|
||||
`openSession` verifies the gateway's reported `gateway_protocol_version` against
|
||||
the version this client was generated for and throws `MxGatewayException` on a
|
||||
mismatch, so an incompatible client fails fast with a clear message instead of
|
||||
issuing commands that fail downstream. A gateway that does not populate the
|
||||
field is accepted unchanged.
|
||||
|
||||
`MxGatewaySession` implements `AutoCloseable`. The try-with-resources `close()`
|
||||
performs a `CloseSession` network RPC but swallows (and logs) any failure of
|
||||
that RPC so a close-time error never replaces the exception a try-with-resources
|
||||
body is already propagating. Call `closeRaw()` explicitly when you need to
|
||||
observe the close result or handle a close-time failure.
|
||||
|
||||
`MxGatewayClient` and `GalaxyRepositoryClient` implement `AutoCloseable`. For a
|
||||
client that owns its channel (built with `connect`), the try-with-resources
|
||||
`close()` shuts the channel down and waits up to the configured connect timeout
|
||||
for termination, forcibly shutting it down on timeout, so in-flight calls and
|
||||
Netty event-loop threads are not left running after the block exits. If the
|
||||
calling thread is interrupted while waiting, the channel is forcibly shut down
|
||||
and the interrupt flag is restored. `closeAndAwaitTermination()` does the same
|
||||
but throws `InterruptedException` for callers that want a checked,
|
||||
blocking-aware shutdown. `close()` is a no-op for a caller-managed channel.
|
||||
|
||||
`MxEventStream` implements `Iterator<MxEvent>` and `AutoCloseable`. Closing it
|
||||
cancels the underlying gRPC stream. Canceling or timing out a Java client call
|
||||
only stops the client from waiting; it does not abort an in-flight MXAccess COM
|
||||
call on the worker STA. The event stream uses gRPC's default auto-inbound flow
|
||||
control with a fixed 16-element buffer and no client-side flow control: this is
|
||||
the gateway's documented fail-fast event-backpressure model, so a consumer that
|
||||
stalls long enough to fill the buffer triggers an overflow that cancels the
|
||||
subscription and surfaces an `MxGatewayException` from the next `next()` call.
|
||||
Drain events promptly and be prepared to resubscribe with a resume cursor.
|
||||
call on the worker STA.
|
||||
|
||||
## Galaxy Repository Browse
|
||||
|
||||
@@ -232,8 +205,8 @@ Create local library and CLI artifacts from `clients/java`:
|
||||
gradle :mxgateway-client:jar :mxgateway-cli:installDist
|
||||
```
|
||||
|
||||
The library jar is under `mxgateway-client/build/libs`. The installed CLI
|
||||
distribution is under `mxgateway-cli/build/install/mxgateway-cli`.
|
||||
The library jar is under `zb-mom-ww-mxgateway-client/build/libs`. The installed CLI
|
||||
distribution is under `zb-mom-ww-mxgateway-cli/build/install/mxgateway-cli`.
|
||||
|
||||
## Integration Checks
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@ ext {
|
||||
}
|
||||
|
||||
subprojects {
|
||||
group = 'com.dohertylan.mxgateway'
|
||||
group = 'com.zb.mom.ww.mxgateway'
|
||||
version = '0.1.0'
|
||||
|
||||
pluginManager.withPlugin('java') {
|
||||
|
||||
-164
@@ -1,164 +0,0 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
|
||||
import com.google.common.util.concurrent.FutureCallback;
|
||||
import com.google.common.util.concurrent.Futures;
|
||||
import com.google.common.util.concurrent.ListenableFuture;
|
||||
import com.google.common.util.concurrent.MoreExecutors;
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
|
||||
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
|
||||
import io.grpc.stub.AbstractStub;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Function;
|
||||
import javax.net.ssl.SSLException;
|
||||
|
||||
/**
|
||||
* Shared channel-builder and future-adaptor helpers used by both
|
||||
* {@link MxGatewayClient} and {@link GalaxyRepositoryClient}.
|
||||
*
|
||||
* <p>Extracted so transport construction, per-call deadlines, and the
|
||||
* {@link ListenableFuture}-to-{@link CompletableFuture} bridge live in one
|
||||
* place instead of being duplicated verbatim across the two clients.
|
||||
*/
|
||||
final class MxGatewayChannels {
|
||||
private MxGatewayChannels() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds a Netty managed channel from the supplied options, applying the
|
||||
* connect timeout, message-size limit, and the configured transport
|
||||
* security mode (plaintext, custom CA trust, or system trust).
|
||||
*
|
||||
* @param options the client options carrying endpoint and transport config
|
||||
* @param tlsErrorPrefix a human-readable prefix for the {@link MxGatewayException}
|
||||
* thrown when a custom CA certificate cannot be loaded
|
||||
* @return a new managed channel; the caller owns its lifecycle
|
||||
*/
|
||||
static ManagedChannel createChannel(MxGatewayClientOptions options, String tlsErrorPrefix) {
|
||||
NettyChannelBuilder builder = NettyChannelBuilder.forTarget(options.endpoint())
|
||||
.maxInboundMessageSize(options.maxGrpcMessageBytes());
|
||||
if (!options.connectTimeout().isNegative()) {
|
||||
builder.withOption(
|
||||
io.grpc.netty.shaded.io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS,
|
||||
Math.toIntExact(options.connectTimeout().toMillis()));
|
||||
}
|
||||
if (options.plaintext()) {
|
||||
builder.usePlaintext();
|
||||
} else if (options.caCertificatePath() != null) {
|
||||
try {
|
||||
builder.sslContext(GrpcSslContexts.forClient()
|
||||
.trustManager(options.caCertificatePath().toFile())
|
||||
.build());
|
||||
} catch (SSLException | RuntimeException error) {
|
||||
// SSLException covers handshake-context failures; RuntimeException
|
||||
// (IllegalArgumentException wrapping CertificateException) covers a
|
||||
// missing or unreadable CA file. Either way callers see one typed
|
||||
// failure instead of a raw, unwrapped exception leaking out.
|
||||
throw new MxGatewayException(tlsErrorPrefix, error);
|
||||
}
|
||||
} else {
|
||||
builder.useTransportSecurity();
|
||||
}
|
||||
if (!options.serverNameOverride().isBlank()) {
|
||||
builder.overrideAuthority(options.serverNameOverride());
|
||||
}
|
||||
return builder.build();
|
||||
}
|
||||
|
||||
/**
|
||||
* Applies the configured per-call deadline to a unary stub.
|
||||
*
|
||||
* @param stub the stub to decorate
|
||||
* @param options the client options carrying the call timeout
|
||||
* @param <T> the concrete stub type
|
||||
* @return the stub with the call deadline applied, or the stub unchanged
|
||||
* when the call timeout is negative (disabled)
|
||||
*/
|
||||
static <T extends AbstractStub<T>> T withDeadline(T stub, MxGatewayClientOptions options) {
|
||||
if (options.callTimeout().isNegative()) {
|
||||
return stub;
|
||||
}
|
||||
return stub.withDeadlineAfter(options.callTimeout().toNanos(), TimeUnit.NANOSECONDS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Applies the configured streaming deadline to a streaming stub.
|
||||
*
|
||||
* @param stub the stub to decorate
|
||||
* @param options the client options carrying the stream timeout
|
||||
* @param <T> the concrete stub type
|
||||
* @return the stub with the stream deadline applied, or the stub unchanged
|
||||
* when the stream timeout is unset or negative (disabled)
|
||||
*/
|
||||
static <T extends AbstractStub<T>> T withStreamDeadline(T stub, MxGatewayClientOptions options) {
|
||||
if (options.streamTimeout() == null || options.streamTimeout().isNegative()) {
|
||||
return stub;
|
||||
}
|
||||
return stub.withDeadlineAfter(options.streamTimeout().toNanos(), TimeUnit.NANOSECONDS);
|
||||
}
|
||||
|
||||
/**
|
||||
* Bridges a Guava {@link ListenableFuture} to a {@link CompletableFuture},
|
||||
* normalising any failure through {@link MxGatewayErrors#fromGrpc} so the
|
||||
* async error surface matches the synchronous methods. Cancelling the
|
||||
* returned future cancels the source RPC.
|
||||
*
|
||||
* @param source the gRPC future-stub result
|
||||
* @param operation the operation name used in normalised error messages
|
||||
* @param <T> the reply type
|
||||
* @return a completable future mirroring the source
|
||||
*/
|
||||
static <T> CompletableFuture<T> toCompletable(ListenableFuture<T> source, String operation) {
|
||||
CompletableFuture<T> target = new CompletableFuture<>();
|
||||
Futures.addCallback(
|
||||
source,
|
||||
new FutureCallback<>() {
|
||||
@Override
|
||||
public void onSuccess(T result) {
|
||||
target.complete(result);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onFailure(Throwable error) {
|
||||
if (error instanceof RuntimeException runtimeException) {
|
||||
target.completeExceptionally(MxGatewayErrors.fromGrpc(operation, runtimeException));
|
||||
return;
|
||||
}
|
||||
target.completeExceptionally(error);
|
||||
}
|
||||
},
|
||||
MoreExecutors.directExecutor());
|
||||
target.whenComplete((ignoredResult, ignoredError) -> {
|
||||
if (target.isCancelled()) {
|
||||
source.cancel(true);
|
||||
}
|
||||
});
|
||||
return target;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adapts a reply-validating function for use inside {@code thenApply} so
|
||||
* any non-{@link MxGatewayException} {@link RuntimeException} it raises is
|
||||
* routed through {@link MxGatewayErrors#fromGrpc}. This keeps the async
|
||||
* error surface consistent with the synchronous methods, which normalise
|
||||
* failures with a {@code try/catch}.
|
||||
*
|
||||
* @param operation the operation name used in normalised error messages
|
||||
* @param validator the validating/transforming function applied to the reply
|
||||
* @param <T> the reply type
|
||||
* @param <R> the result type
|
||||
* @return a function suitable for {@link CompletableFuture#thenApply}
|
||||
*/
|
||||
static <T, R> Function<T, R> normalisingValidator(String operation, Function<T, R> validator) {
|
||||
return reply -> {
|
||||
try {
|
||||
return validator.apply(reply);
|
||||
} catch (MxGatewayException error) {
|
||||
throw error;
|
||||
} catch (RuntimeException error) {
|
||||
throw MxGatewayErrors.fromGrpc(operation, error);
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
-503
@@ -1,503 +0,0 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.Server;
|
||||
import io.grpc.Status;
|
||||
import io.grpc.inprocess.InProcessChannelBuilder;
|
||||
import io.grpc.inprocess.InProcessServerBuilder;
|
||||
import io.grpc.stub.ClientCallStreamObserver;
|
||||
import io.grpc.stub.ClientResponseObserver;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import java.nio.file.Path;
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionException;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import mxaccess_gateway.v1.MxAccessGatewayGrpc;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.AcknowledgeAlarmReply;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.AcknowledgeAlarmRequest;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.ActiveAlarmSnapshot;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.AlarmConditionState;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.MxEvent;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatus;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatusCode;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.QueryActiveAlarmsRequest;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* Regression tests for the Low-severity Client.Java code-review findings
|
||||
* (Client.Java-006 through Client.Java-012). Covers the alarm RPC surface,
|
||||
* async streaming/subscription cancellation, queue overflow, and TLS-config
|
||||
* construction that Client.Java-007 reports as untested.
|
||||
*/
|
||||
final class MxGatewayLowFindingsTests {
|
||||
|
||||
// --- Client.Java-007: AcknowledgeAlarm RPC coverage ---
|
||||
|
||||
@Test
|
||||
void acknowledgeAlarmReturnsReplyAndSendsAuthMetadata() throws Exception {
|
||||
AtomicReference<String> authorization = new AtomicReference<>();
|
||||
AtomicReference<AcknowledgeAlarmRequest> seen = new AtomicReference<>();
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void acknowledgeAlarm(
|
||||
AcknowledgeAlarmRequest request, StreamObserver<AcknowledgeAlarmReply> responseObserver) {
|
||||
seen.set(request);
|
||||
responseObserver.onNext(AcknowledgeAlarmReply.newBuilder()
|
||||
.setSessionId(request.getSessionId())
|
||||
.setProtocolStatus(ok())
|
||||
.setDiagnosticMessage("acked")
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service, "mxgw_keyid_secret", authorization)) {
|
||||
AcknowledgeAlarmReply reply = harness.client().acknowledgeAlarm(AcknowledgeAlarmRequest.newBuilder()
|
||||
.setSessionId("s-1")
|
||||
.setAlarmFullReference("Area1.Pump.PV.HiHi")
|
||||
.setComment("operator note")
|
||||
.build());
|
||||
assertEquals("acked", reply.getDiagnosticMessage());
|
||||
assertEquals("Area1.Pump.PV.HiHi", seen.get().getAlarmFullReference());
|
||||
assertEquals("Bearer mxgw_keyid_secret", authorization.get());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void acknowledgeAlarmThrowsTypedExceptionOnProtocolFailure() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void acknowledgeAlarm(
|
||||
AcknowledgeAlarmRequest request, StreamObserver<AcknowledgeAlarmReply> responseObserver) {
|
||||
responseObserver.onNext(AcknowledgeAlarmReply.newBuilder()
|
||||
.setSessionId(request.getSessionId())
|
||||
.setProtocolStatus(ProtocolStatus.newBuilder()
|
||||
.setCode(ProtocolStatusCode.PROTOCOL_STATUS_CODE_SESSION_NOT_FOUND))
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
assertThrows(
|
||||
MxGatewayException.class,
|
||||
() -> harness.client().acknowledgeAlarm(AcknowledgeAlarmRequest.newBuilder()
|
||||
.setSessionId("missing")
|
||||
.build()));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void acknowledgeAlarmAsyncCompletesWithReply() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void acknowledgeAlarm(
|
||||
AcknowledgeAlarmRequest request, StreamObserver<AcknowledgeAlarmReply> responseObserver) {
|
||||
responseObserver.onNext(AcknowledgeAlarmReply.newBuilder()
|
||||
.setSessionId(request.getSessionId())
|
||||
.setProtocolStatus(ok())
|
||||
.setDiagnosticMessage("async-acked")
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
CompletableFuture<AcknowledgeAlarmReply> future = harness.client()
|
||||
.acknowledgeAlarmAsync(AcknowledgeAlarmRequest.newBuilder().setSessionId("s-2").build());
|
||||
assertEquals("async-acked", future.get(5, TimeUnit.SECONDS).getDiagnosticMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void acknowledgeAlarmAsyncFailsExceptionallyWithTypedException() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void acknowledgeAlarm(
|
||||
AcknowledgeAlarmRequest request, StreamObserver<AcknowledgeAlarmReply> responseObserver) {
|
||||
responseObserver.onError(Status.UNAVAILABLE.withDescription("worker down").asRuntimeException());
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
CompletableFuture<AcknowledgeAlarmReply> future = harness.client()
|
||||
.acknowledgeAlarmAsync(AcknowledgeAlarmRequest.newBuilder().setSessionId("s-3").build());
|
||||
ExecutionException error = assertThrows(
|
||||
ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS));
|
||||
assertTrue(error.getCause() instanceof MxGatewayException, () -> String.valueOf(error.getCause()));
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Java-007: QueryActiveAlarms RPC + subscription coverage ---
|
||||
|
||||
@Test
|
||||
void queryActiveAlarmsDeliversSnapshotsToObserver() throws Exception {
|
||||
ActiveAlarmSnapshot snapshot = ActiveAlarmSnapshot.newBuilder()
|
||||
.setAlarmFullReference("Area1.Tank.Level.Hi")
|
||||
.setSeverity(800)
|
||||
.setCurrentState(AlarmConditionState.ALARM_CONDITION_STATE_ACTIVE)
|
||||
.build();
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void queryActiveAlarms(
|
||||
QueryActiveAlarmsRequest request, StreamObserver<ActiveAlarmSnapshot> responseObserver) {
|
||||
responseObserver.onNext(snapshot);
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
List<ActiveAlarmSnapshot> received = new ArrayList<>();
|
||||
CountDownLatch done = new CountDownLatch(1);
|
||||
harness.client().queryActiveAlarms(
|
||||
QueryActiveAlarmsRequest.newBuilder().setSessionId("s-4").build(),
|
||||
new StreamObserver<>() {
|
||||
@Override
|
||||
public void onNext(ActiveAlarmSnapshot value) {
|
||||
received.add(value);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
done.countDown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
done.countDown();
|
||||
}
|
||||
});
|
||||
assertTrue(done.await(5, TimeUnit.SECONDS), "stream should complete");
|
||||
assertEquals(1, received.size());
|
||||
assertEquals("Area1.Tank.Level.Hi", received.get(0).getAlarmFullReference());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void activeAlarmsSubscriptionCancelBeforeBeforeStartCancelsStream() {
|
||||
MxGatewayActiveAlarmsSubscription subscription = new MxGatewayActiveAlarmsSubscription();
|
||||
ClientResponseObserver<QueryActiveAlarmsRequest, ActiveAlarmSnapshot> observer =
|
||||
subscription.wrap(new StreamObserver<>() {
|
||||
@Override
|
||||
public void onNext(ActiveAlarmSnapshot value) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
}
|
||||
});
|
||||
RecordingActiveAlarmsRequestStream requestStream = new RecordingActiveAlarmsRequestStream();
|
||||
|
||||
subscription.cancel();
|
||||
observer.beforeStart(requestStream);
|
||||
|
||||
assertTrue(requestStream.cancelled);
|
||||
assertEquals("client cancelled active-alarms query", requestStream.cancelMessage);
|
||||
}
|
||||
|
||||
// --- Client.Java-007: async streamEvents + subscription cancellation ---
|
||||
|
||||
@Test
|
||||
void streamEventsAsyncDeliversEventsToObserver() throws Exception {
|
||||
MxEvent event = MxEvent.newBuilder().setWorkerSequence(7).build();
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void streamEvents(StreamEventsRequest request, StreamObserver<MxEvent> responseObserver) {
|
||||
responseObserver.onNext(event);
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
List<MxEvent> received = new ArrayList<>();
|
||||
CountDownLatch done = new CountDownLatch(1);
|
||||
harness.client().streamEventsAsync(
|
||||
StreamEventsRequest.newBuilder().setSessionId("s-5").build(),
|
||||
new StreamObserver<>() {
|
||||
@Override
|
||||
public void onNext(MxEvent value) {
|
||||
received.add(value);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
done.countDown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
done.countDown();
|
||||
}
|
||||
});
|
||||
assertTrue(done.await(5, TimeUnit.SECONDS), "stream should complete");
|
||||
assertEquals(1, received.size());
|
||||
assertEquals(7, received.get(0).getWorkerSequence());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void eventSubscriptionCancelBeforeBeforeStartCancelsStream() {
|
||||
MxGatewayEventSubscription subscription = new MxGatewayEventSubscription();
|
||||
ClientResponseObserver<StreamEventsRequest, MxEvent> observer =
|
||||
subscription.wrap(new StreamObserver<>() {
|
||||
@Override
|
||||
public void onNext(MxEvent value) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
}
|
||||
});
|
||||
RecordingEventsRequestStream requestStream = new RecordingEventsRequestStream();
|
||||
|
||||
subscription.cancel();
|
||||
observer.beforeStart(requestStream);
|
||||
|
||||
assertTrue(requestStream.cancelled);
|
||||
assertEquals("client cancelled event stream", requestStream.cancelMessage);
|
||||
}
|
||||
|
||||
// --- Client.Java-007 / Client.Java-011: MxEventStream queue overflow ---
|
||||
|
||||
@Test
|
||||
void eventStreamQueueOverflowSurfacesExceptionFromNext() {
|
||||
MxEventStream stream = new MxEventStream(2);
|
||||
ClientResponseObserver<StreamEventsRequest, MxEvent> observer = stream.observer();
|
||||
RecordingEventsRequestStream requestStream = new RecordingEventsRequestStream();
|
||||
observer.beforeStart(requestStream);
|
||||
|
||||
// Push far more events than the capacity-2 buffer can hold without draining.
|
||||
for (int i = 0; i < 16; i++) {
|
||||
observer.onNext(MxEvent.newBuilder().setWorkerSequence(i).build());
|
||||
}
|
||||
|
||||
// Overflow must cancel the gRPC call and surface as MxGatewayException.
|
||||
assertTrue(requestStream.cancelled, "overflow should cancel the underlying call");
|
||||
MxGatewayException error = assertThrows(MxGatewayException.class, () -> {
|
||||
while (stream.hasNext()) {
|
||||
stream.next();
|
||||
}
|
||||
});
|
||||
assertTrue(error.getMessage().contains("overflow"), error::getMessage);
|
||||
}
|
||||
|
||||
// --- Client.Java-007: TLS channel construction ---
|
||||
|
||||
@Test
|
||||
void connectWithMissingCaCertificateThrowsTypedTlsException() {
|
||||
MxGatewayClientOptions options = MxGatewayClientOptions.builder()
|
||||
.endpoint("localhost:5001")
|
||||
.apiKey("mxgw_id_secret")
|
||||
.plaintext(false)
|
||||
.caCertificatePath(Path.of("does-not-exist-" + UUID.randomUUID() + ".pem"))
|
||||
.build();
|
||||
|
||||
MxGatewayException error = assertThrows(MxGatewayException.class, () -> MxGatewayClient.connect(options));
|
||||
assertTrue(error.getMessage().contains("TLS"), error::getMessage);
|
||||
|
||||
MxGatewayException galaxyError =
|
||||
assertThrows(MxGatewayException.class, () -> GalaxyRepositoryClient.connect(options));
|
||||
assertTrue(galaxyError.getMessage().contains("TLS"), galaxyError::getMessage);
|
||||
}
|
||||
|
||||
@Test
|
||||
void connectWithSystemTrustBuildsTlsChannelWithoutError() {
|
||||
// No CA path and plaintext=false exercises the useTransportSecurity() branch.
|
||||
MxGatewayClientOptions options = MxGatewayClientOptions.builder()
|
||||
.endpoint("localhost:5001")
|
||||
.apiKey("mxgw_id_secret")
|
||||
.plaintext(false)
|
||||
.build();
|
||||
|
||||
try (MxGatewayClient client = MxGatewayClient.connect(options)) {
|
||||
assertNotNull(client);
|
||||
}
|
||||
try (GalaxyRepositoryClient galaxy = GalaxyRepositoryClient.connect(options)) {
|
||||
assertNotNull(galaxy);
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Java-008: async error surface is normalised ---
|
||||
|
||||
@Test
|
||||
void openSessionAsyncNormalisesNonGatewayRuntimeExceptionFromValidator() {
|
||||
// ensureGatewayProtocolCompatible already throws MxGatewayException; this verifies
|
||||
// the normalisingValidator wrapper routes a stray RuntimeException through fromGrpc.
|
||||
CompletableFuture<String> source = new CompletableFuture<>();
|
||||
CompletableFuture<String> wrapped =
|
||||
source.thenApply(MxGatewayChannels.normalisingValidator("open session", reply -> {
|
||||
throw new IllegalStateException("malformed reply");
|
||||
}));
|
||||
source.complete("payload");
|
||||
|
||||
CompletionException error = assertThrows(CompletionException.class, wrapped::join);
|
||||
assertTrue(error.getCause() instanceof MxGatewayException, () -> String.valueOf(error.getCause()));
|
||||
}
|
||||
|
||||
private static ProtocolStatus ok() {
|
||||
return ProtocolStatus.newBuilder()
|
||||
.setCode(ProtocolStatusCode.PROTOCOL_STATUS_CODE_OK)
|
||||
.build();
|
||||
}
|
||||
|
||||
private static class TestService extends MxAccessGatewayGrpc.MxAccessGatewayImplBase {
|
||||
}
|
||||
|
||||
private record Harness(Server server, ManagedChannel channel, MxGatewayClient client) implements AutoCloseable {
|
||||
static Harness start(MxAccessGatewayGrpc.MxAccessGatewayImplBase service) throws Exception {
|
||||
return start(service, "", new AtomicReference<>());
|
||||
}
|
||||
|
||||
static Harness start(
|
||||
MxAccessGatewayGrpc.MxAccessGatewayImplBase service,
|
||||
String apiKey,
|
||||
AtomicReference<String> authorization)
|
||||
throws Exception {
|
||||
String name = "mxgw-low-" + UUID.randomUUID();
|
||||
io.grpc.ServerInterceptor interceptor = new io.grpc.ServerInterceptor() {
|
||||
@Override
|
||||
public <ReqT, RespT> io.grpc.ServerCall.Listener<ReqT> interceptCall(
|
||||
io.grpc.ServerCall<ReqT, RespT> call,
|
||||
io.grpc.Metadata headers,
|
||||
io.grpc.ServerCallHandler<ReqT, RespT> next) {
|
||||
authorization.set(headers.get(MxGatewayAuthInterceptor.AUTHORIZATION_HEADER));
|
||||
return next.startCall(call, headers);
|
||||
}
|
||||
};
|
||||
Server server = InProcessServerBuilder.forName(name)
|
||||
.directExecutor()
|
||||
.addService(io.grpc.ServerInterceptors.intercept(service, interceptor))
|
||||
.build()
|
||||
.start();
|
||||
ManagedChannel channel = InProcessChannelBuilder.forName(name).directExecutor().build();
|
||||
MxGatewayClient client = new MxGatewayClient(
|
||||
channel,
|
||||
MxGatewayClientOptions.builder()
|
||||
.endpoint("in-process")
|
||||
.apiKey(apiKey)
|
||||
.plaintext(true)
|
||||
.callTimeout(Duration.ofSeconds(5))
|
||||
.streamTimeout(Duration.ofSeconds(5))
|
||||
.build());
|
||||
return new Harness(server, channel, client);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
channel.shutdownNow();
|
||||
server.shutdownNow();
|
||||
}
|
||||
}
|
||||
|
||||
private static final class RecordingEventsRequestStream
|
||||
extends ClientCallStreamObserver<StreamEventsRequest> {
|
||||
private boolean cancelled;
|
||||
private String cancelMessage;
|
||||
|
||||
@Override
|
||||
public void cancel(String message, Throwable cause) {
|
||||
cancelled = true;
|
||||
cancelMessage = message;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isReady() {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setOnReadyHandler(Runnable onReadyHandler) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void request(int count) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setMessageCompression(boolean enable) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void disableAutoInboundFlowControl() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(StreamEventsRequest value) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
}
|
||||
}
|
||||
|
||||
private static final class RecordingActiveAlarmsRequestStream
|
||||
extends ClientCallStreamObserver<QueryActiveAlarmsRequest> {
|
||||
private boolean cancelled;
|
||||
private String cancelMessage;
|
||||
|
||||
@Override
|
||||
public void cancel(String message, Throwable cause) {
|
||||
cancelled = true;
|
||||
cancelMessage = message;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isReady() {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setOnReadyHandler(Runnable onReadyHandler) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void request(int count) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setMessageCompression(boolean enable) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void disableAutoInboundFlowControl() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(QueryActiveAlarmsRequest value) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
}
|
||||
}
|
||||
}
|
||||
-394
@@ -1,394 +0,0 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.Server;
|
||||
import io.grpc.inprocess.InProcessChannelBuilder;
|
||||
import io.grpc.inprocess.InProcessServerBuilder;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import java.time.Duration;
|
||||
import java.util.UUID;
|
||||
import mxaccess_gateway.v1.MxAccessGatewayGrpc;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.CloseSessionReply;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.CloseSessionRequest;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.MxCommandKind;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.MxCommandReply;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.MxCommandRequest;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.OpenSessionReply;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.OpenSessionRequest;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatus;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatusCode;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* Regression tests for the Medium-severity Client.Java code-review findings
|
||||
* (Client.Java-001 through Client.Java-005).
|
||||
*/
|
||||
final class MxGatewayMediumFindingsTests {
|
||||
|
||||
// --- Client.Java-001: redactApiKey must not leak trailing secret chars ---
|
||||
|
||||
@Test
|
||||
void redactApiKeyDoesNotLeakAnyCharacterOfTheSecret() {
|
||||
// mxgw_<key-id>_<secret> — the secret is the segment after the second underscore.
|
||||
String apiKey = "mxgw_keyid01_supersecretvalue";
|
||||
String redacted = MxGatewaySecrets.redactApiKey(apiKey);
|
||||
|
||||
// None of the secret characters may appear in the redacted output.
|
||||
assertFalse(redacted.contains("value"), () -> "redacted form leaked secret tail: " + redacted);
|
||||
assertFalse(redacted.endsWith("alue"), () -> "redacted form leaked trailing secret chars: " + redacted);
|
||||
assertFalse(redacted.contains("supersecret"), () -> "redacted form leaked secret: " + redacted);
|
||||
// The non-secret key-id prefix may stay so the value is still comparable in logs.
|
||||
assertTrue(redacted.startsWith("mxgw_keyid01_"), () -> "redacted form lost key-id prefix: " + redacted);
|
||||
}
|
||||
|
||||
@Test
|
||||
void redactApiKeyForNonGatewayShapedKeyRevealsNothing() {
|
||||
String redacted = MxGatewaySecrets.redactApiKey("plain-opaque-token-1234");
|
||||
assertFalse(redacted.contains("1234"), () -> "redacted form leaked trailing chars: " + redacted);
|
||||
assertFalse(redacted.contains("plain-opaque-token"), () -> "redacted form leaked body: " + redacted);
|
||||
}
|
||||
|
||||
@Test
|
||||
void redactApiKeyStillHandlesNullAndShortInput() {
|
||||
assertEquals("", MxGatewaySecrets.redactApiKey(null));
|
||||
assertEquals("", MxGatewaySecrets.redactApiKey(""));
|
||||
assertEquals("<redacted>", MxGatewaySecrets.redactApiKey("short"));
|
||||
}
|
||||
|
||||
// --- Client.Java-002: terminal-state transition must be deterministic ---
|
||||
|
||||
@Test
|
||||
void eventStreamOverflowExceptionSurvivesASubsequentClose() {
|
||||
// Deterministic reproduction of Client.Java-002: an overflow enqueues the
|
||||
// overflow exception, then a later close() must NOT discard it. The first
|
||||
// terminal condition (overflow) must win and stay observable by next().
|
||||
MxEventStream stream = new MxEventStream(2);
|
||||
io.grpc.stub.ClientResponseObserver<
|
||||
mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest,
|
||||
mxaccess_gateway.v1.MxaccessGateway.MxEvent>
|
||||
observer = stream.observer();
|
||||
observer.beforeStart(new NoopRequestStream());
|
||||
|
||||
// Force a queue overflow on a capacity-2 stream.
|
||||
for (int i = 0; i < 8; i++) {
|
||||
observer.onNext(testEvent(i));
|
||||
}
|
||||
|
||||
// A close() arriving after the overflow must not erase the overflow signal.
|
||||
stream.close();
|
||||
|
||||
MxGatewayException error = assertThrows(MxGatewayException.class, () -> {
|
||||
while (stream.hasNext()) {
|
||||
stream.next();
|
||||
}
|
||||
});
|
||||
assertTrue(error.getMessage().contains("overflow"), error::getMessage);
|
||||
}
|
||||
|
||||
@Test
|
||||
void eventStreamConcurrentOverflowAndCloseAlwaysTerminate() throws Exception {
|
||||
// The terminal-state transition must be serialised: whatever the interleaving
|
||||
// of overflow and close, hasNext() always reaches a terminal state.
|
||||
for (int iteration = 0; iteration < 300; iteration++) {
|
||||
MxEventStream stream = new MxEventStream(2);
|
||||
io.grpc.stub.ClientResponseObserver<
|
||||
mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest,
|
||||
mxaccess_gateway.v1.MxaccessGateway.MxEvent>
|
||||
observer = stream.observer();
|
||||
observer.beforeStart(new NoopRequestStream());
|
||||
|
||||
Thread filler = new Thread(() -> {
|
||||
for (int i = 0; i < 8; i++) {
|
||||
observer.onNext(testEvent(i));
|
||||
}
|
||||
});
|
||||
Thread closer = new Thread(stream::close);
|
||||
filler.start();
|
||||
closer.start();
|
||||
filler.join();
|
||||
closer.join();
|
||||
|
||||
try {
|
||||
while (stream.hasNext()) {
|
||||
stream.next();
|
||||
}
|
||||
} catch (MxGatewayException expected) {
|
||||
assertTrue(expected.getMessage().contains("overflow"), expected::getMessage);
|
||||
}
|
||||
assertFalse(stream.hasNext());
|
||||
}
|
||||
}
|
||||
|
||||
private static final class NoopRequestStream
|
||||
extends io.grpc.stub.ClientCallStreamObserver<mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest> {
|
||||
@Override
|
||||
public void cancel(String message, Throwable cause) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isReady() {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setOnReadyHandler(Runnable onReadyHandler) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void request(int count) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setMessageCompression(boolean enable) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void disableAutoInboundFlowControl() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onNext(mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest value) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Java-003: gateway protocol version mismatch must be rejected ---
|
||||
|
||||
@Test
|
||||
void openSessionRejectsIncompatibleGatewayProtocolVersion() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void openSession(OpenSessionRequest request, StreamObserver<OpenSessionReply> responseObserver) {
|
||||
responseObserver.onNext(OpenSessionReply.newBuilder()
|
||||
.setSessionId("session-mismatch")
|
||||
.setGatewayProtocolVersion(MxGatewayClientVersion.gatewayProtocolVersion() + 1)
|
||||
.setProtocolStatus(ok())
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
MxGatewayException error = assertThrows(
|
||||
MxGatewayException.class,
|
||||
() -> harness.client().openSession("junit-session"));
|
||||
assertTrue(error.getMessage().contains("protocol version"), error::getMessage);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void openSessionAcceptsMatchingOrUnsetGatewayProtocolVersion() throws Exception {
|
||||
TestService matching = new TestService() {
|
||||
@Override
|
||||
public void openSession(OpenSessionRequest request, StreamObserver<OpenSessionReply> responseObserver) {
|
||||
responseObserver.onNext(OpenSessionReply.newBuilder()
|
||||
.setSessionId("session-ok")
|
||||
.setGatewayProtocolVersion(MxGatewayClientVersion.gatewayProtocolVersion())
|
||||
.setProtocolStatus(ok())
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
try (Harness harness = Harness.start(matching)) {
|
||||
assertEquals("session-ok", harness.client().openSession("junit-session").sessionId());
|
||||
}
|
||||
|
||||
// A gateway that leaves the field unset (0) must not be rejected — older gateways
|
||||
// simply do not populate it.
|
||||
TestService unset = new TestService();
|
||||
try (Harness harness = Harness.start(unset)) {
|
||||
assertEquals("session-java", harness.client().openSession("junit-session").sessionId());
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Java-004: missing typed payload AND missing return_value must throw ---
|
||||
|
||||
@Test
|
||||
void registerThrowsWhenReplyHasNeitherTypedPayloadNorReturnValue() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void invoke(MxCommandRequest request, StreamObserver<MxCommandReply> responseObserver) {
|
||||
// Reply with neither register payload nor return_value set.
|
||||
responseObserver.onNext(MxCommandReply.newBuilder()
|
||||
.setSessionId(request.getSessionId())
|
||||
.setKind(request.getCommand().getKind())
|
||||
.setProtocolStatus(ok())
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
MxGatewaySession session = MxGatewaySession.forSessionId(harness.client(), "s");
|
||||
MxGatewayException error = assertThrows(
|
||||
MxGatewayException.class, () -> session.register("c"));
|
||||
assertTrue(error.getMessage().contains("register"), error::getMessage);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void addItemThrowsWhenReplyHasNeitherTypedPayloadNorReturnValue() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void invoke(MxCommandRequest request, StreamObserver<MxCommandReply> responseObserver) {
|
||||
responseObserver.onNext(MxCommandReply.newBuilder()
|
||||
.setSessionId(request.getSessionId())
|
||||
.setKind(request.getCommand().getKind())
|
||||
.setProtocolStatus(ok())
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
MxGatewaySession session = MxGatewaySession.forSessionId(harness.client(), "s");
|
||||
assertThrows(MxGatewayException.class, () -> session.addItem(1, "Tag"));
|
||||
assertThrows(MxGatewayException.class, () -> session.addItem2(1, "Tag", "ctx"));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void addItemStillHonoursReturnValueFallback() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void invoke(MxCommandRequest request, StreamObserver<MxCommandReply> responseObserver) {
|
||||
responseObserver.onNext(MxCommandReply.newBuilder()
|
||||
.setSessionId(request.getSessionId())
|
||||
.setKind(request.getCommand().getKind())
|
||||
.setProtocolStatus(ok())
|
||||
.setReturnValue(mxaccess_gateway.v1.MxaccessGateway.MxValue.newBuilder()
|
||||
.setInt32Value(99))
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
MxGatewaySession session = MxGatewaySession.forSessionId(harness.client(), "s");
|
||||
assertEquals(99, session.addItem(1, "Tag"));
|
||||
}
|
||||
}
|
||||
|
||||
// --- Client.Java-005: close() must not mask the primary try-with-resources error ---
|
||||
|
||||
@Test
|
||||
void closeSuppressesCloseTimeFailureInsteadOfMaskingBodyException() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void closeSession(CloseSessionRequest request, StreamObserver<CloseSessionReply> responseObserver) {
|
||||
responseObserver.onError(io.grpc.Status.UNAVAILABLE
|
||||
.withDescription("WORKER_UNAVAILABLE")
|
||||
.asRuntimeException());
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
IllegalStateException bodyError = assertThrows(IllegalStateException.class, () -> {
|
||||
try (MxGatewaySession session = MxGatewaySession.forSessionId(harness.client(), "s")) {
|
||||
throw new IllegalStateException("body failure");
|
||||
}
|
||||
});
|
||||
// The body exception must propagate; the close-time RPC failure must not replace it.
|
||||
assertEquals("body failure", bodyError.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void closeRawStillSurfacesCloseTimeFailureForCallersWhoWantIt() throws Exception {
|
||||
TestService service = new TestService() {
|
||||
@Override
|
||||
public void closeSession(CloseSessionRequest request, StreamObserver<CloseSessionReply> responseObserver) {
|
||||
responseObserver.onError(io.grpc.Status.UNAVAILABLE
|
||||
.withDescription("WORKER_UNAVAILABLE")
|
||||
.asRuntimeException());
|
||||
}
|
||||
};
|
||||
|
||||
try (Harness harness = Harness.start(service)) {
|
||||
MxGatewaySession session = MxGatewaySession.forSessionId(harness.client(), "s");
|
||||
assertThrows(MxGatewayException.class, session::closeRaw);
|
||||
}
|
||||
}
|
||||
|
||||
private static mxaccess_gateway.v1.MxaccessGateway.MxEvent testEvent(int sequence) {
|
||||
return mxaccess_gateway.v1.MxaccessGateway.MxEvent.newBuilder()
|
||||
.setWorkerSequence(sequence)
|
||||
.build();
|
||||
}
|
||||
|
||||
private static ProtocolStatus ok() {
|
||||
return ProtocolStatus.newBuilder()
|
||||
.setCode(ProtocolStatusCode.PROTOCOL_STATUS_CODE_OK)
|
||||
.build();
|
||||
}
|
||||
|
||||
private static class TestService extends MxAccessGatewayGrpc.MxAccessGatewayImplBase {
|
||||
@Override
|
||||
public void openSession(OpenSessionRequest request, StreamObserver<OpenSessionReply> responseObserver) {
|
||||
responseObserver.onNext(OpenSessionReply.newBuilder()
|
||||
.setSessionId("session-java")
|
||||
.setProtocolStatus(ok())
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void closeSession(CloseSessionRequest request, StreamObserver<CloseSessionReply> responseObserver) {
|
||||
responseObserver.onNext(CloseSessionReply.newBuilder()
|
||||
.setSessionId(request.getSessionId())
|
||||
.setProtocolStatus(ok())
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void invoke(MxCommandRequest request, StreamObserver<MxCommandReply> responseObserver) {
|
||||
responseObserver.onNext(MxCommandReply.newBuilder()
|
||||
.setSessionId(request.getSessionId())
|
||||
.setKind(MxCommandKind.MX_COMMAND_KIND_UNSPECIFIED)
|
||||
.setProtocolStatus(ok())
|
||||
.build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
}
|
||||
|
||||
private record Harness(Server server, ManagedChannel channel, MxGatewayClient client) implements AutoCloseable {
|
||||
static Harness start(MxAccessGatewayGrpc.MxAccessGatewayImplBase service) throws Exception {
|
||||
String name = "mxgw-medium-" + UUID.randomUUID();
|
||||
Server server = InProcessServerBuilder.forName(name)
|
||||
.directExecutor()
|
||||
.addService(service)
|
||||
.build()
|
||||
.start();
|
||||
ManagedChannel channel = InProcessChannelBuilder.forName(name).directExecutor().build();
|
||||
MxGatewayClient client = new MxGatewayClient(
|
||||
channel,
|
||||
MxGatewayClientOptions.builder()
|
||||
.endpoint("in-process")
|
||||
.apiKey("")
|
||||
.plaintext(true)
|
||||
.callTimeout(Duration.ofSeconds(5))
|
||||
.build());
|
||||
return new Harness(server, channel, client);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
channel.shutdownNow();
|
||||
server.shutdownNow();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -16,7 +16,7 @@ dependencyResolutionManagement {
|
||||
}
|
||||
}
|
||||
|
||||
rootProject.name = 'mxaccessgw-java'
|
||||
rootProject.name = 'zb-mom-ww-mxaccessgw-java'
|
||||
|
||||
include 'mxgateway-client'
|
||||
include 'mxgateway-cli'
|
||||
include 'zb-mom-ww-mxgateway-client'
|
||||
include 'zb-mom-ww-mxgateway-cli'
|
||||
|
||||
+133
-1
@@ -170,6 +170,37 @@ public final class MxAccessGatewayGrpc {
|
||||
return getAcknowledgeAlarmMethod;
|
||||
}
|
||||
|
||||
private static volatile io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest,
|
||||
mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage> getStreamAlarmsMethod;
|
||||
|
||||
@io.grpc.stub.annotations.RpcMethod(
|
||||
fullMethodName = SERVICE_NAME + '/' + "StreamAlarms",
|
||||
requestType = mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest.class,
|
||||
responseType = mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage.class,
|
||||
methodType = io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING)
|
||||
public static io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest,
|
||||
mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage> getStreamAlarmsMethod() {
|
||||
io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest, mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage> getStreamAlarmsMethod;
|
||||
if ((getStreamAlarmsMethod = MxAccessGatewayGrpc.getStreamAlarmsMethod) == null) {
|
||||
synchronized (MxAccessGatewayGrpc.class) {
|
||||
if ((getStreamAlarmsMethod = MxAccessGatewayGrpc.getStreamAlarmsMethod) == null) {
|
||||
MxAccessGatewayGrpc.getStreamAlarmsMethod = getStreamAlarmsMethod =
|
||||
io.grpc.MethodDescriptor.<mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest, mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage>newBuilder()
|
||||
.setType(io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING)
|
||||
.setFullMethodName(generateFullMethodName(SERVICE_NAME, "StreamAlarms"))
|
||||
.setSampledToLocalTracing(true)
|
||||
.setRequestMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
|
||||
mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest.getDefaultInstance()))
|
||||
.setResponseMarshaller(io.grpc.protobuf.ProtoUtils.marshaller(
|
||||
mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage.getDefaultInstance()))
|
||||
.setSchemaDescriptor(new MxAccessGatewayMethodDescriptorSupplier("StreamAlarms"))
|
||||
.build();
|
||||
}
|
||||
}
|
||||
}
|
||||
return getStreamAlarmsMethod;
|
||||
}
|
||||
|
||||
private static volatile io.grpc.MethodDescriptor<mxaccess_gateway.v1.MxaccessGateway.QueryActiveAlarmsRequest,
|
||||
mxaccess_gateway.v1.MxaccessGateway.ActiveAlarmSnapshot> getQueryActiveAlarmsMethod;
|
||||
|
||||
@@ -303,6 +334,27 @@ public final class MxAccessGatewayGrpc {
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Session-less central alarm feed. The stream opens with the current
|
||||
* active-alarm snapshot (one `active_alarm` per alarm), then a single
|
||||
* `snapshot_complete`, then a `transition` for every subsequent change.
|
||||
* Served by the gateway's always-on alarm monitor; any number of clients
|
||||
* fan out from the single monitor without opening a worker session.
|
||||
* </pre>
|
||||
*/
|
||||
default void streamAlarms(mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest request,
|
||||
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage> responseObserver) {
|
||||
io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall(getStreamAlarmsMethod(), responseObserver);
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Point-in-time snapshot of the currently-active alarm set served from the
|
||||
* gateway's always-on alarm monitor cache (session-less). Used after a
|
||||
* reconnect to seed Part 9 client state, or to reconcile alarms that may
|
||||
* have been missed during a transport blip. Streamed so callers can
|
||||
* begin processing without buffering the full set.
|
||||
* </pre>
|
||||
*/
|
||||
default void queryActiveAlarms(mxaccess_gateway.v1.MxaccessGateway.QueryActiveAlarmsRequest request,
|
||||
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.ActiveAlarmSnapshot> responseObserver) {
|
||||
@@ -384,6 +436,28 @@ public final class MxAccessGatewayGrpc {
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Session-less central alarm feed. The stream opens with the current
|
||||
* active-alarm snapshot (one `active_alarm` per alarm), then a single
|
||||
* `snapshot_complete`, then a `transition` for every subsequent change.
|
||||
* Served by the gateway's always-on alarm monitor; any number of clients
|
||||
* fan out from the single monitor without opening a worker session.
|
||||
* </pre>
|
||||
*/
|
||||
public void streamAlarms(mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest request,
|
||||
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage> responseObserver) {
|
||||
io.grpc.stub.ClientCalls.asyncServerStreamingCall(
|
||||
getChannel().newCall(getStreamAlarmsMethod(), getCallOptions()), request, responseObserver);
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Point-in-time snapshot of the currently-active alarm set served from the
|
||||
* gateway's always-on alarm monitor cache (session-less). Used after a
|
||||
* reconnect to seed Part 9 client state, or to reconcile alarms that may
|
||||
* have been missed during a transport blip. Streamed so callers can
|
||||
* begin processing without buffering the full set.
|
||||
* </pre>
|
||||
*/
|
||||
public void queryActiveAlarms(mxaccess_gateway.v1.MxaccessGateway.QueryActiveAlarmsRequest request,
|
||||
io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.ActiveAlarmSnapshot> responseObserver) {
|
||||
@@ -449,6 +523,29 @@ public final class MxAccessGatewayGrpc {
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Session-less central alarm feed. The stream opens with the current
|
||||
* active-alarm snapshot (one `active_alarm` per alarm), then a single
|
||||
* `snapshot_complete`, then a `transition` for every subsequent change.
|
||||
* Served by the gateway's always-on alarm monitor; any number of clients
|
||||
* fan out from the single monitor without opening a worker session.
|
||||
* </pre>
|
||||
*/
|
||||
@io.grpc.ExperimentalApi("https://github.com/grpc/grpc-java/issues/10918")
|
||||
public io.grpc.stub.BlockingClientCall<?, mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage>
|
||||
streamAlarms(mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest request) {
|
||||
return io.grpc.stub.ClientCalls.blockingV2ServerStreamingCall(
|
||||
getChannel(), getStreamAlarmsMethod(), getCallOptions(), request);
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Point-in-time snapshot of the currently-active alarm set served from the
|
||||
* gateway's always-on alarm monitor cache (session-less). Used after a
|
||||
* reconnect to seed Part 9 client state, or to reconcile alarms that may
|
||||
* have been missed during a transport blip. Streamed so callers can
|
||||
* begin processing without buffering the full set.
|
||||
* </pre>
|
||||
*/
|
||||
@io.grpc.ExperimentalApi("https://github.com/grpc/grpc-java/issues/10918")
|
||||
public io.grpc.stub.BlockingClientCall<?, mxaccess_gateway.v1.MxaccessGateway.ActiveAlarmSnapshot>
|
||||
@@ -514,6 +611,28 @@ public final class MxAccessGatewayGrpc {
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Session-less central alarm feed. The stream opens with the current
|
||||
* active-alarm snapshot (one `active_alarm` per alarm), then a single
|
||||
* `snapshot_complete`, then a `transition` for every subsequent change.
|
||||
* Served by the gateway's always-on alarm monitor; any number of clients
|
||||
* fan out from the single monitor without opening a worker session.
|
||||
* </pre>
|
||||
*/
|
||||
public java.util.Iterator<mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage> streamAlarms(
|
||||
mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest request) {
|
||||
return io.grpc.stub.ClientCalls.blockingServerStreamingCall(
|
||||
getChannel(), getStreamAlarmsMethod(), getCallOptions(), request);
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Point-in-time snapshot of the currently-active alarm set served from the
|
||||
* gateway's always-on alarm monitor cache (session-less). Used after a
|
||||
* reconnect to seed Part 9 client state, or to reconcile alarms that may
|
||||
* have been missed during a transport blip. Streamed so callers can
|
||||
* begin processing without buffering the full set.
|
||||
* </pre>
|
||||
*/
|
||||
public java.util.Iterator<mxaccess_gateway.v1.MxaccessGateway.ActiveAlarmSnapshot> queryActiveAlarms(
|
||||
mxaccess_gateway.v1.MxaccessGateway.QueryActiveAlarmsRequest request) {
|
||||
@@ -579,7 +698,8 @@ public final class MxAccessGatewayGrpc {
|
||||
private static final int METHODID_INVOKE = 2;
|
||||
private static final int METHODID_STREAM_EVENTS = 3;
|
||||
private static final int METHODID_ACKNOWLEDGE_ALARM = 4;
|
||||
private static final int METHODID_QUERY_ACTIVE_ALARMS = 5;
|
||||
private static final int METHODID_STREAM_ALARMS = 5;
|
||||
private static final int METHODID_QUERY_ACTIVE_ALARMS = 6;
|
||||
|
||||
private static final class MethodHandlers<Req, Resp> implements
|
||||
io.grpc.stub.ServerCalls.UnaryMethod<Req, Resp>,
|
||||
@@ -618,6 +738,10 @@ public final class MxAccessGatewayGrpc {
|
||||
serviceImpl.acknowledgeAlarm((mxaccess_gateway.v1.MxaccessGateway.AcknowledgeAlarmRequest) request,
|
||||
(io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.AcknowledgeAlarmReply>) responseObserver);
|
||||
break;
|
||||
case METHODID_STREAM_ALARMS:
|
||||
serviceImpl.streamAlarms((mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest) request,
|
||||
(io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage>) responseObserver);
|
||||
break;
|
||||
case METHODID_QUERY_ACTIVE_ALARMS:
|
||||
serviceImpl.queryActiveAlarms((mxaccess_gateway.v1.MxaccessGateway.QueryActiveAlarmsRequest) request,
|
||||
(io.grpc.stub.StreamObserver<mxaccess_gateway.v1.MxaccessGateway.ActiveAlarmSnapshot>) responseObserver);
|
||||
@@ -675,6 +799,13 @@ public final class MxAccessGatewayGrpc {
|
||||
mxaccess_gateway.v1.MxaccessGateway.AcknowledgeAlarmRequest,
|
||||
mxaccess_gateway.v1.MxaccessGateway.AcknowledgeAlarmReply>(
|
||||
service, METHODID_ACKNOWLEDGE_ALARM)))
|
||||
.addMethod(
|
||||
getStreamAlarmsMethod(),
|
||||
io.grpc.stub.ServerCalls.asyncServerStreamingCall(
|
||||
new MethodHandlers<
|
||||
mxaccess_gateway.v1.MxaccessGateway.StreamAlarmsRequest,
|
||||
mxaccess_gateway.v1.MxaccessGateway.AlarmFeedMessage>(
|
||||
service, METHODID_STREAM_ALARMS)))
|
||||
.addMethod(
|
||||
getQueryActiveAlarmsMethod(),
|
||||
io.grpc.stub.ServerCalls.asyncServerStreamingCall(
|
||||
@@ -735,6 +866,7 @@ public final class MxAccessGatewayGrpc {
|
||||
.addMethod(getInvokeMethod())
|
||||
.addMethod(getStreamEventsMethod())
|
||||
.addMethod(getAcknowledgeAlarmMethod())
|
||||
.addMethod(getStreamAlarmsMethod())
|
||||
.addMethod(getQueryActiveAlarmsMethod())
|
||||
.build();
|
||||
}
|
||||
|
||||
+152
-2
@@ -8976,17 +8976,36 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
getFullTagReferenceBytes();
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL `dbo.data_type` identifier, passed through unchanged.
|
||||
* This is NOT a member of `mxaccess_gateway.v1.MxDataType` — Galaxy's
|
||||
* type enumeration is distinct from MXAccess's wire data-type enum and
|
||||
* the two must not be cast or compared. The GalaxyRepository service is
|
||||
* metadata-only and deliberately does not share types with
|
||||
* mxaccess_gateway.proto. See docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_data_type = 3;</code>
|
||||
* @return The mxDataType.
|
||||
*/
|
||||
int getMxDataType();
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @return The dataTypeName.
|
||||
*/
|
||||
java.lang.String getDataTypeName();
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @return The bytes for dataTypeName.
|
||||
*/
|
||||
@@ -9012,12 +9031,24 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
boolean getArrayDimensionPresent();
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL attribute-category identifier, passed through unchanged.
|
||||
* Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_attribute_category = 8;</code>
|
||||
* @return The mxAttributeCategory.
|
||||
*/
|
||||
int getMxAttributeCategory();
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL security-classification identifier, passed through
|
||||
* unchanged. Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 security_classification = 9;</code>
|
||||
* @return The securityClassification.
|
||||
*/
|
||||
@@ -9156,6 +9187,15 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
public static final int MX_DATA_TYPE_FIELD_NUMBER = 3;
|
||||
private int mxDataType_ = 0;
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL `dbo.data_type` identifier, passed through unchanged.
|
||||
* This is NOT a member of `mxaccess_gateway.v1.MxDataType` — Galaxy's
|
||||
* type enumeration is distinct from MXAccess's wire data-type enum and
|
||||
* the two must not be cast or compared. The GalaxyRepository service is
|
||||
* metadata-only and deliberately does not share types with
|
||||
* mxaccess_gateway.proto. See docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_data_type = 3;</code>
|
||||
* @return The mxDataType.
|
||||
*/
|
||||
@@ -9168,6 +9208,11 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
@SuppressWarnings("serial")
|
||||
private volatile java.lang.Object dataTypeName_ = "";
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @return The dataTypeName.
|
||||
*/
|
||||
@@ -9185,6 +9230,11 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
}
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @return The bytes for dataTypeName.
|
||||
*/
|
||||
@@ -9239,6 +9289,12 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
public static final int MX_ATTRIBUTE_CATEGORY_FIELD_NUMBER = 8;
|
||||
private int mxAttributeCategory_ = 0;
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL attribute-category identifier, passed through unchanged.
|
||||
* Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_attribute_category = 8;</code>
|
||||
* @return The mxAttributeCategory.
|
||||
*/
|
||||
@@ -9250,6 +9306,12 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
public static final int SECURITY_CLASSIFICATION_FIELD_NUMBER = 9;
|
||||
private int securityClassification_ = 0;
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL security-classification identifier, passed through
|
||||
* unchanged. Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 security_classification = 9;</code>
|
||||
* @return The securityClassification.
|
||||
*/
|
||||
@@ -9956,6 +10018,15 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
|
||||
private int mxDataType_ ;
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL `dbo.data_type` identifier, passed through unchanged.
|
||||
* This is NOT a member of `mxaccess_gateway.v1.MxDataType` — Galaxy's
|
||||
* type enumeration is distinct from MXAccess's wire data-type enum and
|
||||
* the two must not be cast or compared. The GalaxyRepository service is
|
||||
* metadata-only and deliberately does not share types with
|
||||
* mxaccess_gateway.proto. See docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_data_type = 3;</code>
|
||||
* @return The mxDataType.
|
||||
*/
|
||||
@@ -9964,6 +10035,15 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
return mxDataType_;
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL `dbo.data_type` identifier, passed through unchanged.
|
||||
* This is NOT a member of `mxaccess_gateway.v1.MxDataType` — Galaxy's
|
||||
* type enumeration is distinct from MXAccess's wire data-type enum and
|
||||
* the two must not be cast or compared. The GalaxyRepository service is
|
||||
* metadata-only and deliberately does not share types with
|
||||
* mxaccess_gateway.proto. See docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_data_type = 3;</code>
|
||||
* @param value The mxDataType to set.
|
||||
* @return This builder for chaining.
|
||||
@@ -9976,6 +10056,15 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
return this;
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL `dbo.data_type` identifier, passed through unchanged.
|
||||
* This is NOT a member of `mxaccess_gateway.v1.MxDataType` — Galaxy's
|
||||
* type enumeration is distinct from MXAccess's wire data-type enum and
|
||||
* the two must not be cast or compared. The GalaxyRepository service is
|
||||
* metadata-only and deliberately does not share types with
|
||||
* mxaccess_gateway.proto. See docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_data_type = 3;</code>
|
||||
* @return This builder for chaining.
|
||||
*/
|
||||
@@ -9988,6 +10077,11 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
|
||||
private java.lang.Object dataTypeName_ = "";
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @return The dataTypeName.
|
||||
*/
|
||||
@@ -10004,6 +10098,11 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
}
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @return The bytes for dataTypeName.
|
||||
*/
|
||||
@@ -10021,6 +10120,11 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
}
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @param value The dataTypeName to set.
|
||||
* @return This builder for chaining.
|
||||
@@ -10034,6 +10138,11 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
return this;
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @return This builder for chaining.
|
||||
*/
|
||||
@@ -10044,6 +10153,11 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
return this;
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Human-readable name from Galaxy's `dbo.data_type` table (e.g. "Float",
|
||||
* "Integer", "Boolean"). Free-form Galaxy text; not a stable enum.
|
||||
* </pre>
|
||||
*
|
||||
* <code>string data_type_name = 4;</code>
|
||||
* @param value The bytes for dataTypeName to set.
|
||||
* @return This builder for chaining.
|
||||
@@ -10156,6 +10270,12 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
|
||||
private int mxAttributeCategory_ ;
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL attribute-category identifier, passed through unchanged.
|
||||
* Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_attribute_category = 8;</code>
|
||||
* @return The mxAttributeCategory.
|
||||
*/
|
||||
@@ -10164,6 +10284,12 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
return mxAttributeCategory_;
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL attribute-category identifier, passed through unchanged.
|
||||
* Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_attribute_category = 8;</code>
|
||||
* @param value The mxAttributeCategory to set.
|
||||
* @return This builder for chaining.
|
||||
@@ -10176,6 +10302,12 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
return this;
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL attribute-category identifier, passed through unchanged.
|
||||
* Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 mx_attribute_category = 8;</code>
|
||||
* @return This builder for chaining.
|
||||
*/
|
||||
@@ -10188,6 +10320,12 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
|
||||
private int securityClassification_ ;
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL security-classification identifier, passed through
|
||||
* unchanged. Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 security_classification = 9;</code>
|
||||
* @return The securityClassification.
|
||||
*/
|
||||
@@ -10196,6 +10334,12 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
return securityClassification_;
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL security-classification identifier, passed through
|
||||
* unchanged. Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 security_classification = 9;</code>
|
||||
* @param value The securityClassification to set.
|
||||
* @return This builder for chaining.
|
||||
@@ -10208,6 +10352,12 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
return this;
|
||||
}
|
||||
/**
|
||||
* <pre>
|
||||
* Raw Galaxy SQL security-classification identifier, passed through
|
||||
* unchanged. Galaxy-specific; not mapped to any gateway enum. See
|
||||
* docs/GalaxyRepository.md.
|
||||
* </pre>
|
||||
*
|
||||
* <code>int32 security_classification = 9;</code>
|
||||
* @return This builder for chaining.
|
||||
*/
|
||||
@@ -10446,8 +10596,8 @@ public final class GalaxyRepositoryOuterClass extends com.google.protobuf.Genera
|
||||
"sitory.v1.DiscoverHierarchyReply\022h\n\021Watc" +
|
||||
"hDeployEvents\022..galaxy_repository.v1.Wat" +
|
||||
"chDeployEventsRequest\032!.galaxy_repositor" +
|
||||
"y.v1.DeployEvent0\001B#\252\002 MxGateway.Contrac" +
|
||||
"ts.Proto.Galaxyb\006proto3"
|
||||
"y.v1.DeployEvent0\001B-\252\002*ZB.MOM.WW.MxGatew" +
|
||||
"ay.Contracts.Proto.Galaxyb\006proto3"
|
||||
};
|
||||
descriptor = com.google.protobuf.Descriptors.FileDescriptor
|
||||
.internalBuildGeneratedFileFrom(descriptorData,
|
||||
|
||||
+17396
-1143
File diff suppressed because it is too large
Load Diff
@@ -12608,8 +12608,8 @@ public final class MxaccessWorker extends com.google.protobuf.GeneratedFile {
|
||||
"CONVERSION_FAILED\020\010\022\"\n\036WORKER_FAULT_CATE" +
|
||||
"GORY_STA_HUNG\020\t\022(\n$WORKER_FAULT_CATEGORY" +
|
||||
"_QUEUE_OVERFLOW\020\n\022*\n&WORKER_FAULT_CATEGO" +
|
||||
"RY_SHUTDOWN_TIMEOUT\020\013B\034\252\002\031MxGateway.Cont" +
|
||||
"racts.Protob\006proto3"
|
||||
"RY_SHUTDOWN_TIMEOUT\020\013B&\252\002#ZB.MOM.WW.MxGa" +
|
||||
"teway.Contracts.Protob\006proto3"
|
||||
};
|
||||
descriptor = com.google.protobuf.Descriptors.FileDescriptor
|
||||
.internalBuildGeneratedFileFrom(descriptorData,
|
||||
|
||||
+2
-2
@@ -3,11 +3,11 @@ plugins {
|
||||
}
|
||||
|
||||
dependencies {
|
||||
implementation project(':mxgateway-client')
|
||||
implementation project(':zb-mom-ww-mxgateway-client')
|
||||
implementation "com.google.protobuf:protobuf-java-util:${protobufVersion}"
|
||||
implementation "info.picocli:picocli:${picocliVersion}"
|
||||
}
|
||||
|
||||
application {
|
||||
mainClass = 'com.dohertylan.mxgateway.cli.MxGatewayCli'
|
||||
mainClass = 'com.zb.mom.ww.mxgateway.cli.MxGatewayCli'
|
||||
}
|
||||
+21
-48
@@ -1,14 +1,14 @@
|
||||
package com.dohertylan.mxgateway.cli;
|
||||
package com.zb.mom.ww.mxgateway.cli;
|
||||
|
||||
import com.dohertylan.mxgateway.client.DeployEventStream;
|
||||
import com.dohertylan.mxgateway.client.GalaxyRepositoryClient;
|
||||
import com.dohertylan.mxgateway.client.MxEventStream;
|
||||
import com.dohertylan.mxgateway.client.MxGatewayClient;
|
||||
import com.dohertylan.mxgateway.client.MxGatewayClientOptions;
|
||||
import com.dohertylan.mxgateway.client.MxGatewayClientVersion;
|
||||
import com.dohertylan.mxgateway.client.MxGatewaySecrets;
|
||||
import com.dohertylan.mxgateway.client.MxGatewaySession;
|
||||
import com.dohertylan.mxgateway.client.MxValues;
|
||||
import com.zb.mom.ww.mxgateway.client.DeployEventStream;
|
||||
import com.zb.mom.ww.mxgateway.client.GalaxyRepositoryClient;
|
||||
import com.zb.mom.ww.mxgateway.client.MxEventStream;
|
||||
import com.zb.mom.ww.mxgateway.client.MxGatewayClient;
|
||||
import com.zb.mom.ww.mxgateway.client.MxGatewayClientOptions;
|
||||
import com.zb.mom.ww.mxgateway.client.MxGatewayClientVersion;
|
||||
import com.zb.mom.ww.mxgateway.client.MxGatewaySecrets;
|
||||
import com.zb.mom.ww.mxgateway.client.MxGatewaySession;
|
||||
import com.zb.mom.ww.mxgateway.client.MxValues;
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.DeployEvent;
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.GalaxyAttribute;
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.GalaxyObject;
|
||||
@@ -661,60 +661,33 @@ public final class MxGatewayCli implements Callable<Integer> {
|
||||
@Option(names = "--timeout", defaultValue = "30s", description = "Per-call timeout.")
|
||||
String timeout;
|
||||
|
||||
/**
|
||||
* Returns this options object unchanged.
|
||||
*
|
||||
* <p>Retained as a no-op for call sites that read more naturally as
|
||||
* {@code common.resolved()}. Resolution of the API key and timeout is
|
||||
* computed lazily on demand by {@link #resolvedApiKey()} and
|
||||
* {@link #resolvedTimeout()}, so {@link #toClientOptions()} and
|
||||
* {@link #redactedJsonMap()} produce correct output regardless of
|
||||
* whether this method was ever called.
|
||||
*
|
||||
* @return this options object
|
||||
*/
|
||||
private String resolvedApiKey = "";
|
||||
private Duration resolvedTimeout = Duration.ofSeconds(30);
|
||||
|
||||
CommonOptions resolved() {
|
||||
resolvedApiKey = apiKey == null || apiKey.isBlank() ? System.getenv(apiKeyEnv) : apiKey;
|
||||
if (resolvedApiKey == null) {
|
||||
resolvedApiKey = "";
|
||||
}
|
||||
resolvedTimeout = parseDuration(timeout);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves the effective API key: the explicit {@code --api-key} value
|
||||
* when non-blank, otherwise the value of the {@code --api-key-env}
|
||||
* environment variable, otherwise an empty string. Computed on each
|
||||
* call so there is no stale cached state.
|
||||
*
|
||||
* @return the resolved API key, never {@code null}
|
||||
*/
|
||||
String resolvedApiKey() {
|
||||
String resolved = apiKey == null || apiKey.isBlank() ? System.getenv(apiKeyEnv) : apiKey;
|
||||
return resolved == null ? "" : resolved;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves the effective per-call timeout from the {@code --timeout}
|
||||
* option. Computed on each call so there is no stale cached state.
|
||||
*
|
||||
* @return the resolved call timeout
|
||||
*/
|
||||
Duration resolvedTimeout() {
|
||||
return parseDuration(timeout);
|
||||
}
|
||||
|
||||
MxGatewayClientOptions toClientOptions() {
|
||||
return MxGatewayClientOptions.builder()
|
||||
.endpoint(endpoint)
|
||||
.apiKey(resolvedApiKey())
|
||||
.apiKey(resolvedApiKey)
|
||||
.plaintext(plaintext)
|
||||
.caCertificatePath(caFile)
|
||||
.serverNameOverride(serverNameOverride)
|
||||
.callTimeout(resolvedTimeout())
|
||||
.callTimeout(resolvedTimeout)
|
||||
.build();
|
||||
}
|
||||
|
||||
Map<String, Object> redactedJsonMap() {
|
||||
Map<String, Object> values = new LinkedHashMap<>();
|
||||
values.put("endpoint", endpoint);
|
||||
values.put("apiKey", MxGatewaySecrets.redactApiKey(resolvedApiKey()));
|
||||
values.put("apiKey", MxGatewaySecrets.redactApiKey(resolvedApiKey));
|
||||
values.put("apiKeyEnv", apiKeyEnv);
|
||||
values.put("plaintext", plaintext);
|
||||
values.put("caFile", caFile == null ? "" : caFile.toString());
|
||||
+3
-5
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.cli;
|
||||
package com.zb.mom.ww.mxgateway.cli;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
@@ -62,10 +62,8 @@ final class MxGatewayCliTests {
|
||||
assertEquals(0, run.exitCode());
|
||||
assertTrue(run.output().contains("\"command\":\"open-session\""));
|
||||
assertTrue(run.output().contains("\"sessionId\":\"session-cli\""));
|
||||
// Only the non-secret mxgw_<key-id>_ prefix survives; the secret is fully masked.
|
||||
assertTrue(run.output().contains("mxgw_visible_***"));
|
||||
assertTrue(run.output().contains("mxgw***********cret"));
|
||||
assertFalse(run.output().contains("visible_secret"));
|
||||
assertFalse(run.output().contains("cret"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -298,7 +296,7 @@ final class MxGatewayCliTests {
|
||||
}
|
||||
|
||||
@Override
|
||||
public com.dohertylan.mxgateway.client.MxEventStream streamEventsAfter(long afterWorkerSequence) {
|
||||
public com.zb.mom.ww.mxgateway.client.MxEventStream streamEventsAfter(long afterWorkerSequence) {
|
||||
throw new UnsupportedOperationException("stream-events is covered by client tests");
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -22,7 +22,7 @@ dependencies {
|
||||
sourceSets {
|
||||
main {
|
||||
proto {
|
||||
srcDir rootProject.file('../../src/MxGateway.Contracts/Protos')
|
||||
srcDir rootProject.file('../../src/ZB.MOM.WW.MxGateway.Contracts/Protos')
|
||||
include 'mxaccess_gateway.proto'
|
||||
include 'mxaccess_worker.proto'
|
||||
include 'galaxy_repository.proto'
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.DeployEvent;
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.WatchDeployEventsRequest;
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.DeployEvent;
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.WatchDeployEventsRequest;
|
||||
+97
-52
@@ -1,5 +1,8 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import com.google.common.util.concurrent.FutureCallback;
|
||||
import com.google.common.util.concurrent.Futures;
|
||||
import com.google.common.util.concurrent.MoreExecutors;
|
||||
import galaxy_repository.v1.GalaxyRepositoryGrpc;
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.DeployEvent;
|
||||
import galaxy_repository.v1.GalaxyRepositoryOuterClass.DiscoverHierarchyReply;
|
||||
@@ -14,6 +17,8 @@ import com.google.protobuf.Timestamp;
|
||||
import io.grpc.Channel;
|
||||
import io.grpc.ClientInterceptors;
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
|
||||
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import java.time.Instant;
|
||||
import java.util.Iterator;
|
||||
@@ -22,6 +27,7 @@ import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import javax.net.ssl.SSLException;
|
||||
|
||||
/**
|
||||
* Thin wrapper around the generated {@link GalaxyRepositoryGrpc} stubs that
|
||||
@@ -72,8 +78,7 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
* @return a connected client
|
||||
*/
|
||||
public static GalaxyRepositoryClient connect(MxGatewayClientOptions options) {
|
||||
return new GalaxyRepositoryClient(
|
||||
MxGatewayChannels.createChannel(options, "failed to configure galaxy repository TLS"), options);
|
||||
return new GalaxyRepositoryClient(createChannel(options), options);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -82,7 +87,7 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
* @return the blocking stub
|
||||
*/
|
||||
public GalaxyRepositoryGrpc.GalaxyRepositoryBlockingStub rawBlockingStub() {
|
||||
return MxGatewayChannels.withDeadline(blockingStub, options);
|
||||
return withDeadline(blockingStub);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -91,7 +96,7 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
* @return the future stub
|
||||
*/
|
||||
public GalaxyRepositoryGrpc.GalaxyRepositoryFutureStub rawFutureStub() {
|
||||
return MxGatewayChannels.withDeadline(futureStub, options);
|
||||
return withDeadline(futureStub);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -128,9 +133,7 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
* exceptionally with {@link MxGatewayException} on failure
|
||||
*/
|
||||
public CompletableFuture<Boolean> testConnectionAsync() {
|
||||
return MxGatewayChannels.toCompletable(
|
||||
rawFutureStub().testConnection(TestConnectionRequest.getDefaultInstance()),
|
||||
"galaxy test connection")
|
||||
return toCompletable(rawFutureStub().testConnection(TestConnectionRequest.getDefaultInstance()))
|
||||
.thenApply(TestConnectionReply::getOk);
|
||||
}
|
||||
|
||||
@@ -162,11 +165,8 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
* completed exceptionally with {@link MxGatewayException} on failure
|
||||
*/
|
||||
public CompletableFuture<Optional<Instant>> getLastDeployTimeAsync() {
|
||||
return MxGatewayChannels.toCompletable(
|
||||
rawFutureStub().getLastDeployTime(GetLastDeployTimeRequest.getDefaultInstance()),
|
||||
"galaxy get last deploy time")
|
||||
.thenApply(MxGatewayChannels.normalisingValidator(
|
||||
"galaxy get last deploy time", GalaxyRepositoryClient::mapDeployTime));
|
||||
return toCompletable(rawFutureStub().getLastDeployTime(GetLastDeployTimeRequest.getDefaultInstance()))
|
||||
.thenApply(GalaxyRepositoryClient::mapDeployTime);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -224,8 +224,7 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
*/
|
||||
public DeployEventStream watchDeployEvents(Instant lastSeenDeployTime) {
|
||||
DeployEventStream stream = new DeployEventStream(16);
|
||||
MxGatewayChannels.withStreamDeadline(rawAsyncStub(), options)
|
||||
.watchDeployEvents(buildWatchRequest(lastSeenDeployTime), stream.observer());
|
||||
withStreamDeadline(rawAsyncStub()).watchDeployEvents(buildWatchRequest(lastSeenDeployTime), stream.observer());
|
||||
return stream;
|
||||
}
|
||||
|
||||
@@ -254,7 +253,7 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
Instant lastSeenDeployTime, StreamObserver<DeployEvent> observer) {
|
||||
Objects.requireNonNull(observer, "observer");
|
||||
DeployEventSubscription subscription = new DeployEventSubscription();
|
||||
MxGatewayChannels.withStreamDeadline(rawAsyncStub(), options)
|
||||
withStreamDeadline(rawAsyncStub())
|
||||
.watchDeployEvents(buildWatchRequest(lastSeenDeployTime), subscription.wrap(observer));
|
||||
return subscription;
|
||||
}
|
||||
@@ -270,31 +269,17 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
return builder.build();
|
||||
}
|
||||
|
||||
/**
|
||||
* Shuts the owned channel down and awaits termination so try-with-resources
|
||||
* callers do not leave in-flight calls or Netty event-loop threads running
|
||||
* after the block exits.
|
||||
*
|
||||
* <p>Waits up to the configured connect timeout for graceful termination
|
||||
* and forcibly shuts the channel down on timeout. If the calling thread is
|
||||
* interrupted while waiting, the channel is forcibly shut down and the
|
||||
* thread's interrupt flag is restored. No-op for clients that do not own
|
||||
* their channel. For an explicitly checked, blocking-aware shutdown call
|
||||
* {@link #closeAndAwaitTermination()}.
|
||||
*/
|
||||
private <T extends io.grpc.stub.AbstractStub<T>> T withStreamDeadline(T stub) {
|
||||
if (options.streamTimeout() == null || options.streamTimeout().isNegative()) {
|
||||
return stub;
|
||||
}
|
||||
return stub.withDeadlineAfter(options.streamTimeout().toNanos(), TimeUnit.NANOSECONDS);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
if (ownedChannel == null) {
|
||||
return;
|
||||
}
|
||||
ownedChannel.shutdown();
|
||||
try {
|
||||
if (!ownedChannel.awaitTermination(options.connectTimeout().toMillis(), TimeUnit.MILLISECONDS)) {
|
||||
ownedChannel.shutdownNow();
|
||||
}
|
||||
} catch (InterruptedException error) {
|
||||
ownedChannel.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
if (ownedChannel != null) {
|
||||
ownedChannel.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -322,26 +307,86 @@ public final class GalaxyRepositoryClient implements AutoCloseable {
|
||||
return Optional.of(Instant.ofEpochSecond(ts.getSeconds(), ts.getNanos()));
|
||||
}
|
||||
|
||||
private static ManagedChannel createChannel(MxGatewayClientOptions options) {
|
||||
NettyChannelBuilder builder = NettyChannelBuilder.forTarget(options.endpoint())
|
||||
.maxInboundMessageSize(options.maxGrpcMessageBytes());
|
||||
if (!options.connectTimeout().isNegative()) {
|
||||
builder.withOption(
|
||||
io.grpc.netty.shaded.io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS,
|
||||
Math.toIntExact(options.connectTimeout().toMillis()));
|
||||
}
|
||||
if (options.plaintext()) {
|
||||
builder.usePlaintext();
|
||||
} else if (options.caCertificatePath() != null) {
|
||||
try {
|
||||
builder.sslContext(GrpcSslContexts.forClient()
|
||||
.trustManager(options.caCertificatePath().toFile())
|
||||
.build());
|
||||
} catch (SSLException error) {
|
||||
throw new MxGatewayException("failed to configure galaxy repository TLS", error);
|
||||
}
|
||||
} else {
|
||||
builder.useTransportSecurity();
|
||||
}
|
||||
if (!options.serverNameOverride().isBlank()) {
|
||||
builder.overrideAuthority(options.serverNameOverride());
|
||||
}
|
||||
return builder.build();
|
||||
}
|
||||
|
||||
private <T extends io.grpc.stub.AbstractStub<T>> T withDeadline(T stub) {
|
||||
if (options.callTimeout().isNegative()) {
|
||||
return stub;
|
||||
}
|
||||
return stub.withDeadlineAfter(options.callTimeout().toNanos(), TimeUnit.NANOSECONDS);
|
||||
}
|
||||
|
||||
private CompletableFuture<List<GalaxyObject>> discoverHierarchyPageAsync(
|
||||
String pageToken, java.util.ArrayList<GalaxyObject> objects, java.util.HashSet<String> seenPageTokens) {
|
||||
DiscoverHierarchyRequest request = DiscoverHierarchyRequest.newBuilder()
|
||||
.setPageSize(DISCOVER_HIERARCHY_PAGE_SIZE)
|
||||
.setPageToken(pageToken)
|
||||
.build();
|
||||
return MxGatewayChannels.toCompletable(rawFutureStub().discoverHierarchy(request), "galaxy discover hierarchy")
|
||||
.thenCompose(reply -> {
|
||||
objects.addAll(reply.getObjectsList());
|
||||
if (reply.getNextPageToken().isBlank()) {
|
||||
return CompletableFuture.completedFuture(objects);
|
||||
return toCompletable(rawFutureStub().discoverHierarchy(request)).thenCompose(reply -> {
|
||||
objects.addAll(reply.getObjectsList());
|
||||
if (reply.getNextPageToken().isBlank()) {
|
||||
return CompletableFuture.completedFuture(objects);
|
||||
}
|
||||
if (!seenPageTokens.add(reply.getNextPageToken())) {
|
||||
CompletableFuture<List<GalaxyObject>> failed = new CompletableFuture<>();
|
||||
failed.completeExceptionally(new MxGatewayException(
|
||||
"galaxy discover hierarchy returned repeated page token: " + reply.getNextPageToken()));
|
||||
return failed;
|
||||
}
|
||||
return discoverHierarchyPageAsync(reply.getNextPageToken(), objects, seenPageTokens);
|
||||
});
|
||||
}
|
||||
|
||||
private static <T> CompletableFuture<T> toCompletable(com.google.common.util.concurrent.ListenableFuture<T> source) {
|
||||
CompletableFuture<T> target = new CompletableFuture<>();
|
||||
Futures.addCallback(
|
||||
source,
|
||||
new FutureCallback<>() {
|
||||
@Override
|
||||
public void onSuccess(T result) {
|
||||
target.complete(result);
|
||||
}
|
||||
if (!seenPageTokens.add(reply.getNextPageToken())) {
|
||||
CompletableFuture<List<GalaxyObject>> failed = new CompletableFuture<>();
|
||||
failed.completeExceptionally(new MxGatewayException(
|
||||
"galaxy discover hierarchy returned repeated page token: "
|
||||
+ reply.getNextPageToken()));
|
||||
return failed;
|
||||
|
||||
@Override
|
||||
public void onFailure(Throwable error) {
|
||||
if (error instanceof RuntimeException runtimeException) {
|
||||
target.completeExceptionally(MxGatewayErrors.fromGrpc("galaxy async call", runtimeException));
|
||||
return;
|
||||
}
|
||||
target.completeExceptionally(error);
|
||||
}
|
||||
return discoverHierarchyPageAsync(reply.getNextPageToken(), objects, seenPageTokens);
|
||||
});
|
||||
},
|
||||
MoreExecutors.directExecutor());
|
||||
target.whenComplete((ignoredResult, ignoredError) -> {
|
||||
if (target.isCancelled()) {
|
||||
source.cancel(true);
|
||||
}
|
||||
});
|
||||
return target;
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import mxaccess_gateway.v1.MxaccessGateway.MxCommandReply;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatus;
|
||||
+9
-57
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import io.grpc.Status;
|
||||
import io.grpc.StatusRuntimeException;
|
||||
@@ -21,35 +21,13 @@ import mxaccess_gateway.v1.MxaccessGateway.StreamEventsRequest;
|
||||
* stream cancels the underlying gRPC call. If the queue overflows the call is
|
||||
* cancelled and a follow-up call to {@link #next()} throws
|
||||
* {@link MxGatewayException}.
|
||||
*
|
||||
* <p><strong>Backpressure (fail-fast):</strong> this adaptor relies on gRPC's
|
||||
* default auto-inbound flow control — the async stub auto-requests messages, so
|
||||
* the gateway can push events faster than the consumer drains the bounded
|
||||
* 16-element buffer. There is intentionally <em>no</em> real client flow
|
||||
* control: a consumer that stalls long enough to let the buffer fill triggers
|
||||
* an immediate overflow that cancels the subscription and surfaces an
|
||||
* {@link MxGatewayException} on the next {@link #next()} call. This matches the
|
||||
* gateway's documented fail-fast event-backpressure design — a slow consumer
|
||||
* loses its subscription rather than silently dropping events. Consumers that
|
||||
* cannot keep up must drain {@link #next()} promptly (e.g. hand events to their
|
||||
* own larger queue) and be prepared to resubscribe with a resume cursor.
|
||||
*
|
||||
* <p><strong>Threading:</strong> the iterator methods ({@link #hasNext()} and
|
||||
* {@link #next()}) are <em>not</em> thread-safe and must be driven by a single
|
||||
* consumer thread. {@link #close()} may be called from any thread. Terminal
|
||||
* state transitions (queue overflow, server completion, and {@code close()})
|
||||
* are serialised so that the first terminal condition wins deterministically:
|
||||
* once an overflow exception has been observed it is never silently replaced
|
||||
* by an end-of-stream marker.
|
||||
*/
|
||||
public final class MxEventStream implements Iterator<MxEvent>, AutoCloseable {
|
||||
private static final Object END = new Object();
|
||||
|
||||
private final BlockingQueue<Object> queue;
|
||||
private final Object terminalLock = new Object();
|
||||
private volatile ClientCallStreamObserver<StreamEventsRequest> requestStream;
|
||||
private volatile boolean closed;
|
||||
private boolean terminated;
|
||||
private Object next;
|
||||
|
||||
MxEventStream(int capacity) {
|
||||
@@ -120,7 +98,7 @@ public final class MxEventStream implements Iterator<MxEvent>, AutoCloseable {
|
||||
if (stream != null) {
|
||||
stream.cancel("client cancelled event stream", null);
|
||||
}
|
||||
terminate(null);
|
||||
offer(END);
|
||||
}
|
||||
|
||||
private Object take() {
|
||||
@@ -137,7 +115,10 @@ public final class MxEventStream implements Iterator<MxEvent>, AutoCloseable {
|
||||
private void offer(Object value) {
|
||||
Objects.requireNonNull(value, "value");
|
||||
if (value == END) {
|
||||
terminate(null);
|
||||
if (!queue.offer(value)) {
|
||||
queue.clear();
|
||||
queue.offer(value);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (!queue.offer(value)) {
|
||||
@@ -145,38 +126,9 @@ public final class MxEventStream implements Iterator<MxEvent>, AutoCloseable {
|
||||
if (stream != null) {
|
||||
stream.cancel("client event stream queue overflowed", null);
|
||||
}
|
||||
terminate(new MxGatewayException("gateway stream events queue overflowed"));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Drives the single terminal transition. The first caller wins: a later
|
||||
* end-of-stream or {@code close()} cannot overwrite or discard an overflow
|
||||
* exception that has already been published to the consumer.
|
||||
*
|
||||
* @param fault the fault to surface to the consumer, or {@code null} for a
|
||||
* clean end-of-stream
|
||||
*/
|
||||
private void terminate(MxGatewayException fault) {
|
||||
synchronized (terminalLock) {
|
||||
if (terminated) {
|
||||
return;
|
||||
}
|
||||
terminated = true;
|
||||
if (fault != null) {
|
||||
// Make room for the fault marker; the consumer only needs the
|
||||
// terminal signal, queued data events are no longer relevant.
|
||||
queue.clear();
|
||||
queue.offer(fault);
|
||||
queue.offer(END);
|
||||
return;
|
||||
}
|
||||
// Clean end-of-stream: ensure the END marker is delivered even when
|
||||
// the queue is currently full of undrained data events.
|
||||
if (!queue.offer(END)) {
|
||||
queue.clear();
|
||||
queue.offer(END);
|
||||
}
|
||||
queue.clear();
|
||||
queue.offer(new MxGatewayException("gateway stream events queue overflowed"));
|
||||
queue.offer(END);
|
||||
}
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import io.grpc.stub.ClientCallStreamObserver;
|
||||
import io.grpc.stub.ClientResponseObserver;
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import io.grpc.CallOptions;
|
||||
import io.grpc.Channel;
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
/**
|
||||
* Thrown when the gateway rejects a call because the supplied API key is
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
/**
|
||||
* Thrown when the gateway accepts an API key but rejects a call because the
|
||||
+95
-68
@@ -1,13 +1,19 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import com.google.common.util.concurrent.FutureCallback;
|
||||
import com.google.common.util.concurrent.Futures;
|
||||
import com.google.common.util.concurrent.MoreExecutors;
|
||||
import com.google.protobuf.Duration;
|
||||
import io.grpc.Channel;
|
||||
import io.grpc.ClientInterceptors;
|
||||
import io.grpc.ManagedChannel;
|
||||
import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
|
||||
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import javax.net.ssl.SSLException;
|
||||
import mxaccess_gateway.v1.MxAccessGatewayGrpc;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.AcknowledgeAlarmReply;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.AcknowledgeAlarmRequest;
|
||||
@@ -73,8 +79,7 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
* @return a connected client
|
||||
*/
|
||||
public static MxGatewayClient connect(MxGatewayClientOptions options) {
|
||||
return new MxGatewayClient(
|
||||
MxGatewayChannels.createChannel(options, "failed to configure gateway TLS"), options);
|
||||
return new MxGatewayClient(createChannel(options), options);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -83,7 +88,7 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
* @return the blocking stub
|
||||
*/
|
||||
public MxAccessGatewayGrpc.MxAccessGatewayBlockingStub rawBlockingStub() {
|
||||
return MxGatewayChannels.withDeadline(blockingStub, options);
|
||||
return withDeadline(blockingStub);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -92,7 +97,7 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
* @return the future stub
|
||||
*/
|
||||
public MxAccessGatewayGrpc.MxAccessGatewayFutureStub rawFutureStub() {
|
||||
return MxGatewayChannels.withDeadline(futureStub, options);
|
||||
return withDeadline(futureStub);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -145,7 +150,6 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
try {
|
||||
OpenSessionReply reply = rawBlockingStub().openSession(request);
|
||||
MxGatewayErrors.ensureProtocolSuccess("open session", reply.getProtocolStatus(), null);
|
||||
ensureGatewayProtocolCompatible(reply);
|
||||
return reply;
|
||||
} catch (RuntimeException error) {
|
||||
if (error instanceof MxGatewayException) {
|
||||
@@ -155,24 +159,6 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Verifies that the gateway speaks the protocol version this client was
|
||||
* generated against. A gateway that leaves {@code gateway_protocol_version}
|
||||
* unset (value {@code 0}, e.g. an older gateway) is accepted unchanged.
|
||||
*
|
||||
* @param reply the {@code OpenSessionReply} returned by the gateway
|
||||
* @throws MxGatewayException if the gateway reports an incompatible protocol version
|
||||
*/
|
||||
private static void ensureGatewayProtocolCompatible(OpenSessionReply reply) {
|
||||
int gatewayVersion = reply.getGatewayProtocolVersion();
|
||||
int clientVersion = MxGatewayClientVersion.gatewayProtocolVersion();
|
||||
if (gatewayVersion != 0 && gatewayVersion != clientVersion) {
|
||||
throw new MxGatewayException("gateway protocol version mismatch: gateway reports "
|
||||
+ gatewayVersion + " but this client was built for " + clientVersion
|
||||
+ "; upgrade the client or gateway so the protocol versions match");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Invokes {@code OpenSession} asynchronously.
|
||||
*
|
||||
@@ -181,13 +167,11 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
* with {@link MxGatewayException} on failure
|
||||
*/
|
||||
public CompletableFuture<OpenSessionReply> openSessionAsync(OpenSessionRequest request) {
|
||||
CompletableFuture<OpenSessionReply> future =
|
||||
MxGatewayChannels.toCompletable(rawFutureStub().openSession(request), "open session");
|
||||
return future.thenApply(MxGatewayChannels.normalisingValidator("open session", reply -> {
|
||||
CompletableFuture<OpenSessionReply> future = toCompletable(rawFutureStub().openSession(request));
|
||||
return future.thenApply(reply -> {
|
||||
MxGatewayErrors.ensureProtocolSuccess("open session", reply.getProtocolStatus(), null);
|
||||
ensureGatewayProtocolCompatible(reply);
|
||||
return reply;
|
||||
}));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -222,13 +206,12 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
* on failure
|
||||
*/
|
||||
public CompletableFuture<MxCommandReply> invokeAsync(MxCommandRequest request) {
|
||||
CompletableFuture<MxCommandReply> future =
|
||||
MxGatewayChannels.toCompletable(rawFutureStub().invoke(request), "invoke");
|
||||
return future.thenApply(MxGatewayChannels.normalisingValidator("invoke", reply -> {
|
||||
CompletableFuture<MxCommandReply> future = toCompletable(rawFutureStub().invoke(request));
|
||||
return future.thenApply(reply -> {
|
||||
MxGatewayErrors.ensureProtocolSuccess("invoke", reply.getProtocolStatus(), reply);
|
||||
MxGatewayErrors.ensureMxAccessSuccess("invoke", reply);
|
||||
return reply;
|
||||
}));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -261,7 +244,7 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
*/
|
||||
public MxEventStream streamEvents(StreamEventsRequest request) {
|
||||
MxEventStream stream = new MxEventStream(16);
|
||||
MxGatewayChannels.withStreamDeadline(rawAsyncStub(), options).streamEvents(request, stream.observer());
|
||||
withStreamDeadline(rawAsyncStub()).streamEvents(request, stream.observer());
|
||||
return stream;
|
||||
}
|
||||
|
||||
@@ -276,17 +259,15 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
public MxGatewayEventSubscription streamEventsAsync(
|
||||
StreamEventsRequest request, StreamObserver<MxEvent> observer) {
|
||||
MxGatewayEventSubscription subscription = new MxGatewayEventSubscription();
|
||||
MxGatewayChannels.withStreamDeadline(rawAsyncStub(), options)
|
||||
.streamEvents(request, subscription.wrap(observer));
|
||||
withStreamDeadline(rawAsyncStub()).streamEvents(request, subscription.wrap(observer));
|
||||
return subscription;
|
||||
}
|
||||
|
||||
/**
|
||||
* Acknowledges an active MXAccess alarm condition through the gateway.
|
||||
*
|
||||
* <p>The gateway authorizes this request against the API key's
|
||||
* {@code admin} scope (the gateway scope resolver maps alarm RPCs to the
|
||||
* default {@code admin} scope) and forwards the acknowledge to the
|
||||
* <p>The gateway authenticates the request against the API key's
|
||||
* {@code invoke:alarm-ack} scope and forwards the acknowledge to the
|
||||
* worker's MXAccess session; the resulting native MxStatus is returned
|
||||
* in the reply. Acks are idempotent at the MxAccess layer.
|
||||
*
|
||||
@@ -315,12 +296,11 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
* with {@link MxGatewayException} on failure
|
||||
*/
|
||||
public CompletableFuture<AcknowledgeAlarmReply> acknowledgeAlarmAsync(AcknowledgeAlarmRequest request) {
|
||||
CompletableFuture<AcknowledgeAlarmReply> future =
|
||||
MxGatewayChannels.toCompletable(rawFutureStub().acknowledgeAlarm(request), "acknowledge alarm");
|
||||
return future.thenApply(MxGatewayChannels.normalisingValidator("acknowledge alarm", reply -> {
|
||||
CompletableFuture<AcknowledgeAlarmReply> future = toCompletable(rawFutureStub().acknowledgeAlarm(request));
|
||||
return future.thenApply(reply -> {
|
||||
MxGatewayErrors.ensureProtocolSuccess("acknowledge alarm", reply.getProtocolStatus(), null);
|
||||
return reply;
|
||||
}));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -336,36 +316,14 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
public MxGatewayActiveAlarmsSubscription queryActiveAlarms(
|
||||
QueryActiveAlarmsRequest request, StreamObserver<ActiveAlarmSnapshot> observer) {
|
||||
MxGatewayActiveAlarmsSubscription subscription = new MxGatewayActiveAlarmsSubscription();
|
||||
MxGatewayChannels.withStreamDeadline(rawAsyncStub(), options)
|
||||
.queryActiveAlarms(request, subscription.wrap(observer));
|
||||
withStreamDeadline(rawAsyncStub()).queryActiveAlarms(request, subscription.wrap(observer));
|
||||
return subscription;
|
||||
}
|
||||
|
||||
/**
|
||||
* Shuts the owned channel down and awaits termination so try-with-resources
|
||||
* callers do not leave in-flight calls or Netty event-loop threads running
|
||||
* after the block exits.
|
||||
*
|
||||
* <p>Waits up to the configured connect timeout for graceful termination
|
||||
* and forcibly shuts the channel down on timeout. If the calling thread is
|
||||
* interrupted while waiting, the channel is forcibly shut down and the
|
||||
* thread's interrupt flag is restored. No-op for clients that do not own
|
||||
* their channel. For an explicitly checked, blocking-aware shutdown call
|
||||
* {@link #closeAndAwaitTermination()}.
|
||||
*/
|
||||
@Override
|
||||
public void close() {
|
||||
if (ownedChannel == null) {
|
||||
return;
|
||||
}
|
||||
ownedChannel.shutdown();
|
||||
try {
|
||||
if (!ownedChannel.awaitTermination(options.connectTimeout().toMillis(), TimeUnit.MILLISECONDS)) {
|
||||
ownedChannel.shutdownNow();
|
||||
}
|
||||
} catch (InterruptedException error) {
|
||||
ownedChannel.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
if (ownedChannel != null) {
|
||||
ownedChannel.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -385,6 +343,75 @@ public final class MxGatewayClient implements AutoCloseable {
|
||||
}
|
||||
}
|
||||
|
||||
private static ManagedChannel createChannel(MxGatewayClientOptions options) {
|
||||
NettyChannelBuilder builder = NettyChannelBuilder.forTarget(options.endpoint())
|
||||
.maxInboundMessageSize(options.maxGrpcMessageBytes());
|
||||
if (!options.connectTimeout().isNegative()) {
|
||||
builder.withOption(
|
||||
io.grpc.netty.shaded.io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS,
|
||||
Math.toIntExact(options.connectTimeout().toMillis()));
|
||||
}
|
||||
if (options.plaintext()) {
|
||||
builder.usePlaintext();
|
||||
} else if (options.caCertificatePath() != null) {
|
||||
try {
|
||||
builder.sslContext(GrpcSslContexts.forClient()
|
||||
.trustManager(options.caCertificatePath().toFile())
|
||||
.build());
|
||||
} catch (SSLException error) {
|
||||
throw new MxGatewayException("failed to configure gateway TLS", error);
|
||||
}
|
||||
} else {
|
||||
builder.useTransportSecurity();
|
||||
}
|
||||
if (!options.serverNameOverride().isBlank()) {
|
||||
builder.overrideAuthority(options.serverNameOverride());
|
||||
}
|
||||
return builder.build();
|
||||
}
|
||||
|
||||
private <T extends io.grpc.stub.AbstractStub<T>> T withDeadline(T stub) {
|
||||
if (options.callTimeout().isNegative()) {
|
||||
return stub;
|
||||
}
|
||||
return stub.withDeadlineAfter(options.callTimeout().toNanos(), TimeUnit.NANOSECONDS);
|
||||
}
|
||||
|
||||
private <T extends io.grpc.stub.AbstractStub<T>> T withStreamDeadline(T stub) {
|
||||
if (options.streamTimeout() == null || options.streamTimeout().isNegative()) {
|
||||
return stub;
|
||||
}
|
||||
return stub.withDeadlineAfter(options.streamTimeout().toNanos(), TimeUnit.NANOSECONDS);
|
||||
}
|
||||
|
||||
private static <T> CompletableFuture<T> toCompletable(com.google.common.util.concurrent.ListenableFuture<T> source) {
|
||||
CompletableFuture<T> target = new CompletableFuture<>();
|
||||
Futures.addCallback(
|
||||
source,
|
||||
new FutureCallback<>() {
|
||||
@Override
|
||||
public void onSuccess(T result) {
|
||||
target.complete(result);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onFailure(Throwable error) {
|
||||
if (error instanceof RuntimeException runtimeException) {
|
||||
target.completeExceptionally(MxGatewayErrors.fromGrpc("async call", runtimeException));
|
||||
return;
|
||||
}
|
||||
target.completeExceptionally(error);
|
||||
}
|
||||
},
|
||||
MoreExecutors.directExecutor());
|
||||
target.whenComplete((ignoredResult, ignoredError) -> {
|
||||
if (target.isCancelled()) {
|
||||
source.cancel(true);
|
||||
}
|
||||
});
|
||||
return target;
|
||||
}
|
||||
|
||||
static ProtocolStatusCode okStatusCode() {
|
||||
return ProtocolStatusCode.PROTOCOL_STATUS_CODE_OK;
|
||||
}
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.time.Duration;
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
/**
|
||||
* Reports the client and protocol version numbers compiled into this build.
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package com.dohertylan.mxgateway.client;
|
||||
package com.zb.mom.ww.mxgateway.client;
|
||||
|
||||
import mxaccess_gateway.v1.MxaccessGateway.MxCommandReply;
|
||||
import mxaccess_gateway.v1.MxaccessGateway.ProtocolStatus;
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user