-
-
Notifications
You must be signed in to change notification settings - Fork 480
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix subscription reconstruction after reconnection
- fix a bug in client that causes subscriptions to fail to reconstruct if the number of monitored items to rebuild was greater than the operational limit maxMonitoredItemsPerCall
- Loading branch information
1 parent
b775d57
commit 8817c9b
Showing
9 changed files
with
316 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
125 changes: 125 additions & 0 deletions
125
...s/node-opcua-end2end-test/test/end_to_end/test_e2e_large_subscription_and_reconnection.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| import { AttributeIds, CreateMonitoredItemsResponse, OPCUAClient, OPCUAServer, resolveNodeId, StatusCodes, TimestampsToReturn } from "node-opcua"; | ||
| import "should"; | ||
|
|
||
| async function pause(ms: number) { | ||
| return await new Promise((resolve) => setTimeout(resolve, ms)); | ||
| } | ||
| const describe = require("node-opcua-leak-detector").describeWithLeakDetector; | ||
| describe("[CLIENT] recreating large subscription during reconnection", () => { | ||
|
|
||
| let server: OPCUAServer; | ||
| async function startServer(maxMonitoredItemsPerCall: number) { | ||
| server = new OPCUAServer({ | ||
| serverCapabilities: { | ||
| operationLimits: { | ||
| maxMonitoredItemsPerCall, | ||
| } | ||
| } | ||
| }); | ||
|
|
||
| await server.start(); | ||
| return server; | ||
| } | ||
| async function stopServer(server: OPCUAServer) { | ||
| await server.shutdown(); | ||
| } | ||
|
|
||
| it("recreating large subscription during reconnection should not lead to BadTooManyOperation", async () => { | ||
|
|
||
| const maxMonitoredItemsPerCall = 2; | ||
| // Given a server with a small maxMonitoredItemsPerCall value | ||
| let server = await startServer(maxMonitoredItemsPerCall); | ||
| const endpointUrl = server.getEndpointUrl(); | ||
|
|
||
| // Given a client with a large number of monitoredItem | ||
| const client = OPCUAClient.create({ | ||
| }); | ||
| client.on("backoff", () => console.log("backoff")); | ||
| client.on("connection_failed", () => console.log("connection has failed")); | ||
| client.on("after_reconnection", () => console.log("after reconnection")); | ||
| client.on("connection_lost", () => console.log("connection lost")); | ||
| client.on("after_reconnection", () => console.log("after_reconnection")); | ||
|
|
||
| client.on("send_request", (request) => { | ||
| }); | ||
|
|
||
| let createMonitoredItemsResponses: CreateMonitoredItemsResponse[] = []; | ||
| client.on("receive_response", (response) => { | ||
|
|
||
| if ( | ||
| response.constructor.name === "CreateSubscriptionResponse" || | ||
| response.constructor.name === "CreateMonitoredItemsResponse" | ||
| ) { | ||
| // console.log(response.toString()); | ||
| } | ||
| if (response.constructor.name === "CreateMonitoredItemsResponse") { | ||
| createMonitoredItemsResponses.push(response as CreateMonitoredItemsResponse); | ||
| } | ||
| }); | ||
|
|
||
| await client.connect(endpointUrl); | ||
|
|
||
| const session = await client.createSession(); | ||
| const subscription = await session.createSubscription2({ | ||
| requestedLifetimeCount: 10, | ||
| requestedPublishingInterval: 100, | ||
| requestedMaxKeepAliveCount: 5 | ||
| }); | ||
|
|
||
| const nodeId = resolveNodeId("Server_ServerStatus_CurrentTime"); | ||
|
|
||
| // Given that the client has more monitored Items than maxMonitoredItemsPerCall | ||
| for (let i = 0; i < maxMonitoredItemsPerCall + 1; i++) { | ||
| const m = await subscription.monitor({ | ||
| nodeId, attributeId: AttributeIds.Value | ||
| }, { samplingInterval: 10 }, TimestampsToReturn.Both) | ||
| } | ||
|
|
||
|
|
||
| createMonitoredItemsResponses.length.should.eql(maxMonitoredItemsPerCall + 1); | ||
| createMonitoredItemsResponses = []; | ||
|
|
||
| // When the server stops and restarts | ||
| await stopServer(server); | ||
| await pause(1000); | ||
|
|
||
| let isSessionRestored = false; | ||
| session.on('session_restored', () => { | ||
| isSessionRestored = true | ||
| console.log("Session Restored !"); | ||
| }) | ||
|
|
||
| server = await startServer(maxMonitoredItemsPerCall); | ||
|
|
||
| // wait until reconnection is completedf | ||
| while (session.isReconnecting && !isSessionRestored) { | ||
| await pause(100); | ||
| } | ||
|
|
||
| await pause(100); | ||
| console.log("------------------------------------------------------------") | ||
|
|
||
| await session.close(); | ||
| await client.disconnect(); | ||
| await stopServer(server); | ||
|
|
||
| // verify | ||
| if (createMonitoredItemsResponses.length === 0) { | ||
| throw new Error("createMonitoredItemsResponse missing"); | ||
| } | ||
| let n = 0; | ||
| for (let i = 0; i < createMonitoredItemsResponses.length; i++) { | ||
| createMonitoredItemsResponses[0].responseHeader.serviceResult.should.eql(StatusCodes.Good); | ||
|
|
||
| n += createMonitoredItemsResponses[i].results!.length; | ||
| for (const r of createMonitoredItemsResponses[i].results!) { | ||
| r.statusCode!.should.eql(StatusCodes.Good); | ||
| } | ||
| } | ||
| n.should.eql(maxMonitoredItemsPerCall + 1); | ||
|
|
||
|
|
||
|
|
||
| }); | ||
|
|
||
| }); |
24 changes: 24 additions & 0 deletions
24
packages/node-opcua-pseudo-session/source/basic_session_with_subscription.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| import { | ||
| CreateSubscriptionRequestOptions, | ||
| CreateSubscriptionResponse, | ||
| CreateMonitoredItemsRequestOptions, | ||
| CreateMonitoredItemsResponse, | ||
| } from "node-opcua-service-subscription"; | ||
| import { | ||
| ResponseCallback | ||
| } from "./basic_session_interface"; | ||
|
|
||
| /** | ||
| * @module node-opcua-pseudo-session | ||
| */ | ||
| export interface IBasicSessionWithSubscription { | ||
|
|
||
| createSubscription(options: CreateSubscriptionRequestOptions, callback: ResponseCallback<CreateSubscriptionResponse>): void; | ||
| createSubscription(options: CreateSubscriptionRequestOptions): Promise<CreateSubscriptionResponse>; | ||
|
|
||
| // setMonitoringMode(options: SetMonitoringModeRequestLike, callback: ResponseCallback<SetMonitoringModeResponse>): void; | ||
| // setMonitoringMode(options: SetMonitoringModeRequestLike): Promise<SetMonitoringModeResponse>; | ||
|
|
||
| createMonitoredItems(options: CreateMonitoredItemsRequestOptions, callback: ResponseCallback<CreateMonitoredItemsResponse>): void; | ||
| createMonitoredItems(options: CreateMonitoredItemsRequestOptions): Promise<CreateMonitoredItemsResponse>; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
54 changes: 54 additions & 0 deletions
54
packages/node-opcua-pseudo-session/source/create_monitored_items_limit.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| /** | ||
| * @module node-opcua-pseudo-session | ||
| */ | ||
| import { | ||
| CreateMonitoredItemsRequest, | ||
| CreateMonitoredItemsResponse, | ||
| } from "node-opcua-service-subscription"; | ||
| import { | ||
| IBasicSession | ||
| } from "./basic_session_interface"; | ||
| import { | ||
| IBasicSessionWithSubscription | ||
| } from "./basic_session_with_subscription"; | ||
|
|
||
| export async function createMonitoredItemsLimit( | ||
| maxMonitoredItemsPerCall: number, | ||
| session: IBasicSessionWithSubscription, | ||
| createMonitoredItemsRequest: CreateMonitoredItemsRequest | ||
| ): Promise<CreateMonitoredItemsResponse>; | ||
|
|
||
|
|
||
| export async function createMonitoredItemsLimit( | ||
| maxMonitoredItemsPerCall: number, | ||
| session: IBasicSessionWithSubscription, | ||
| createMonitoredItemsRequest: CreateMonitoredItemsRequest | ||
| ): Promise<CreateMonitoredItemsResponse> { | ||
| const _session2 = session as IBasicSessionWithSubscription; | ||
|
|
||
| if ( | ||
| maxMonitoredItemsPerCall <= 0 || | ||
| !createMonitoredItemsRequest.itemsToCreate || | ||
| createMonitoredItemsRequest.itemsToCreate.length <= maxMonitoredItemsPerCall | ||
| ) { | ||
| return _session2.createMonitoredItems(createMonitoredItemsRequest); | ||
| } | ||
| const n = [...(createMonitoredItemsRequest.itemsToCreate || [])]; | ||
| const response = new CreateMonitoredItemsResponse({ | ||
| diagnosticInfos: null, | ||
| results: [] | ||
| }); | ||
| do { | ||
| const c = n.splice(0, maxMonitoredItemsPerCall); | ||
| const cmi = new CreateMonitoredItemsRequest({ | ||
| subscriptionId: createMonitoredItemsRequest.subscriptionId, | ||
| timestampsToReturn: createMonitoredItemsRequest.timestampsToReturn, | ||
| itemsToCreate: c | ||
| }); | ||
| const r = await _session2.createMonitoredItems(cmi); | ||
| for (const i of r.results!) { | ||
| response.results!.push(i); | ||
| } | ||
| } while (n.length); | ||
| return response; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.