Add filtering option to stream subscription

.github/workflows/go.yml

@@ -0,0 +1,56 @@
+###########################################################################
+# Copyright 2025. Ivan Eroshkin
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+###########################################################################
+
+# This workflow will build a golang project
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-go
+
+name: go-netconf-client
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+jobs:
+  linters:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Go
+      uses: actions/setup-go@v4
+      with:
+        go-version: '1.24'
+
+    - name: Golang vet
+      run: make go-vet
+
+    - name: Linters
+      run: make go-lint
+
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v4
+        with:
+          go-version: '1.24'
+
+      - name: Unit tests
+        run: make go-test

.gitignore

@@ -1 +1,3 @@
-.idea/
+.idea/
+build/
+vendor/

Makefile

@@ -0,0 +1,56 @@
+###########################################################################
+# Copyright 2025. Ivan Eroshkin
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+###########################################################################
+
+PROJECT_DIR=$(shell dirname $(realpath $(firstword $(MAKEFILE_LIST))))
+COVERPROFILE_PATH=$(PROJECT_DIR)/build/coverage.txt
+
+all: help
+
+test-ci: # @HELP runs CI/CD pipeline locally
+test-ci: go-vet go-lint go-test
+
+go-vet: # @HELP examines Go code and reports suspicious constructs
+	go vet ./...
+
+go-lint-install: # @HELP installs linters (i.e., 'golint') locally
+	go install golang.org/x/lint/golint@latest
+
+go-lint: # @HELP runs linters against the Go codebase (except 'vendor' folder)
+go-lint: go-lint-install
+	golint ./...
+
+go-test: # @HELP runs unit tests to test the Go code
+	mkdir -p $(PROJECT_DIR)/build
+	go test -v ./... -race -coverprofile=$(COVERPROFILE_PATH) -covermode=atomic
+
+go-tidy: # @HELP runs go mod commands (i.e., 'go mod tidy && go mod vendor')
+	go mod tidy
+	go mod vendor
+
+clean: # @HELP cleans dependencies
+	rm -r $(PROJECT_DIR)/build/
+	rm -r $(PROJECT_DIR)/vendor/
+
+help:
+	@echo "Following 'make' targets can be invoked:"
+	@grep -E '^.*: *# *@HELP' $(MAKEFILE_LIST) \
+    | sort \
+    | awk ' \
+        BEGIN {FS = ": *# *@HELP"}; \
+        {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}; \
+    '
+	@echo ""
+

examples/test-rpc.go

@@ -7,13 +7,12 @@ import (
 	"os"
 	"time"
 
-	"github.com/openshift-telco/go-netconf-client/netconf"
-	"github.com/openshift-telco/go-netconf-client/netconf/message"
+	"github.com/vitrifi/go-netconf-client/netconf"
+	"github.com/vitrifi/go-netconf-client/netconf/message"
 	"golang.org/x/crypto/ssh"
 )
 
 func main() {
-
 	// java -jar lighty-notifications-device-15.0.1-SNAPSHOT.jar 12345
 	testNotification()
 
@@ -24,7 +23,6 @@ func main() {
 }
 
 func testNotification() {
-
 	notificationSession := createSession(12346)
 
 	callback := func(event netconf.Event) {
@@ -46,7 +44,7 @@ func testNotification() {
 	}
 
 	d := message.NewCloseSession()
-	notificationSession.AsyncRPC(d, defaultLogRpcReplyCallback(d.MessageID))
+	notificationSession.AsyncRPC(d, defaultLogRPCReplyCallback(d.MessageID))
 
 	notificationSession.Listener.Remove(message.NetconfNotificationStreamHandler)
 	notificationSession.Listener.WaitForMessages()
@@ -70,42 +68,41 @@ func testRPC() {
 // Add a 100ms delay after each RPC to leave enough time for the device to reply
 // Else, too many request and things get bad.
 func execRPC(session *netconf.Session) {
-
 	// Get Config
 	g := message.NewGetConfig(message.DatastoreRunning, message.FilterTypeSubtree, "")
-	session.AsyncRPC(g, defaultLogRpcReplyCallback(g.MessageID))
+	session.AsyncRPC(g, defaultLogRPCReplyCallback(g.MessageID))
 	time.Sleep(100 * time.Millisecond)
 
 	// Get
 	gt := message.NewGet("", "")
-	session.AsyncRPC(gt, defaultLogRpcReplyCallback(gt.MessageID))
+	session.AsyncRPC(gt, defaultLogRPCReplyCallback(gt.MessageID))
 	time.Sleep(100 * time.Millisecond)
 
 	// Lock
 	l := message.NewLock(message.DatastoreCandidate)
-	session.AsyncRPC(l, defaultLogRpcReplyCallback(l.MessageID))
+	session.AsyncRPC(l, defaultLogRPCReplyCallback(l.MessageID))
 	time.Sleep(100 * time.Millisecond)
 
 	// EditConfig
 	data := "<toaster xmlns=\"http://netconfcentral.org/ns/toaster\">\n    <darknessFactor>750</darknessFactor>\n</toaster>"
 	e := message.NewEditConfig(message.DatastoreCandidate, message.DefaultOperationTypeMerge, data)
-	session.AsyncRPC(e, defaultLogRpcReplyCallback(e.MessageID))
+	session.AsyncRPC(e, defaultLogRPCReplyCallback(e.MessageID))
 	time.Sleep(100 * time.Millisecond)
 
 	// Commit
-	c := message.NewCommit()
-	session.AsyncRPC(c, defaultLogRpcReplyCallback(c.MessageID))
+	c := message.NewCommit("")
+	session.AsyncRPC(c, defaultLogRPCReplyCallback(c.MessageID))
 	time.Sleep(100 * time.Millisecond)
 
 	// Unlock
 	u := message.NewUnlock(message.DatastoreCandidate)
-	session.AsyncRPC(u, defaultLogRpcReplyCallback(u.MessageID))
+	session.AsyncRPC(u, defaultLogRPCReplyCallback(u.MessageID))
 	time.Sleep(100 * time.Millisecond)
 
 	// RPC
 	d := "    <make-toast xmlns=\"http://netconfcentral.org/ns/toaster\">\n        <toasterDoneness>9</toasterDoneness>\n        <toasterToastType>frozen-waffle</toasterToastType>\n     </make-toast>"
 	rpc := message.NewRPC(d)
-	session.AsyncRPC(rpc, defaultLogRpcReplyCallback(rpc.MessageID))
+	session.AsyncRPC(rpc, defaultLogRPCReplyCallback(rpc.MessageID))
 	time.Sleep(100 * time.Millisecond)
 
 	// RPCs
@@ -124,7 +121,7 @@ func execRPC(session *netconf.Session) {
 
 	// Close Session
 	d2 := message.NewCloseSession()
-	session.AsyncRPC(d2, defaultLogRpcReplyCallback(d2.MessageID))
+	session.AsyncRPC(d2, defaultLogRPCReplyCallback(d2.MessageID))
 
 	session.Listener.WaitForMessages()
 }
@@ -151,13 +148,13 @@ func createSession(port int) *netconf.Session {
 	return s
 }
 
-func defaultLogRpcReplyCallback(eventId string) netconf.Callback {
+func defaultLogRPCReplyCallback(eventID string) netconf.Callback {
 	return func(event netconf.Event) {
 		reply := event.RPCReply()
 		if reply == nil {
 			println("Failed to execute RPC")
 		}
-		if event.EventID() == eventId {
+		if event.EventID() == eventID {
 			println("Successfully executed RPC")
 			println(reply.RawReply)
 		}

go.mod

@@ -1,7 +1,7 @@
-module github.com/openshift-telco/go-netconf-client
+module github.com/vitrifi/go-netconf-client
 
-go 1.22.2
+go 1.23.4
 
-require golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
+require golang.org/x/crypto v0.36.0
 
-require golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e // indirect
+require golang.org/x/sys v0.31.0 // indirect

go.sum

@@ -1,6 +1,6 @@
-golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 h1:7I4JAnoQBe7ZtJcBaYHi5UtiO8tQHbUSXxL+pnGRANg=
-golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e h1:WUoyKPm6nCo1BnNUvPGnFG3T5DUVem42yDJZZ4CNxMA=
-golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34=
+golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc=
+golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
+golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/term v0.30.0 h1:PQ39fJZ+mfadBm0y5WlL4vlM7Sx1Hgf13sMIY2+QS9Y=
+golang.org/x/term v0.30.0/go.mod h1:NYYFdzHoI5wRh/h5tDMdMqCqPJZEuNqVR5xJLd/n67g=

netconf/events.go

@@ -17,7 +17,8 @@ limitations under the License.
 package netconf
 
 import (
-	"github.com/openshift-telco/go-netconf-client/netconf/message"
+	"github.com/vitrifi/go-netconf-client/netconf/message"
+	"sync"
 	"time"
 )
 
@@ -47,20 +48,26 @@ type Callback func(Event)
 // those events occur, dispatch them its according callback functions.
 type Dispatcher struct {
 	callbacks map[string]Callback
+	mtx       sync.Mutex
 }
 
 // init a dispatcher creating the callbacks map.
 func (d *Dispatcher) init() {
 	d.callbacks = make(map[string]Callback)
+	d.mtx = sync.Mutex{}
 }
 
 // Register a callback function for the specified eventID.
 func (d *Dispatcher) Register(eventID string, callback Callback) {
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
 	d.callbacks[eventID] = callback
 }
 
 // Remove a callback function for the specified eventID.
 func (d *Dispatcher) Remove(eventID string) {
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
 	delete(d.callbacks, eventID)
 }
 
@@ -82,7 +89,9 @@ func (d *Dispatcher) Dispatch(eventID string, eventType EventType, value interfa
 	}
 
 	// Dispatch the event to the callback
+	d.mtx.Lock()
 	callback := d.callbacks[eventID]
+	d.mtx.Unlock()
 	if callback == nil {
 		return
 	}

netconf/message/commit.go

@@ -24,9 +24,9 @@ type Commit struct {
 }
 
 // NewCommit can be used to create a `commit` message.
-func NewCommit() *Commit {
+func NewCommit(msg string) *Commit {
 	var rpc Commit
-	rpc.Commit = ""
+	rpc.Commit = msg
 	rpc.MessageID = uuid()
 	return &rpc
 }

netconf/message/discard-changes.go

@@ -0,0 +1,32 @@
+/*
+Copyright 2025. Ivan Eroshkin
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package message
+
+// DiscardChanges represents the NETCONF `discard-changes` message.
+// https://datatracker.ietf.org/doc/html/rfc6241#section-8.3.4.2
+type DiscardChanges struct {
+	RPC
+	DiscardChanges interface{} `xml:"discard-changes"`
+}
+
+// NewDiscardChanges can be used to create a `commit` message.
+func NewDiscardChanges(msg string) *DiscardChanges {
+	var rpc DiscardChanges
+	rpc.DiscardChanges = msg
+	rpc.MessageID = uuid()
+	return &rpc
+}

netconf/message/notification.go

@@ -23,7 +23,8 @@ const (
 	NetconfNotificationXmlns = "urn:ietf:params:xml:ns:netconf:notification:1.0"
 	// NetconfNotificationStreamHandler identifies the callback registration for a `create-subscription`
 	NetconfNotificationStreamHandler = "DEFAULT_NOTIFICATION_STREAM"
-	NotificationMessageRegex         = ".*notification"
+	// NotificationMessageRegex represents a regexp for notification messages.
+	NotificationMessageRegex = ".*notification"
 )
 
 // Notification defines a reply to a Notification
@@ -74,6 +75,7 @@ type CreateSubscription struct {
 type CreateSubscriptionData struct {
 	XMLNS     string `xml:"xmlns,attr"`
 	Stream    string `xml:"stream,omitempty"` // default is NETCONF
+	Filter    string `xml:",innerxml"`
 	StartTime string `xml:"startTime,omitempty"`
 	StopTime  string `xml:"stopTime,omitempty"`
 }
@@ -82,18 +84,18 @@ type CreateSubscriptionData struct {
 func NewCreateSubscriptionDefault() *CreateSubscription {
 	var rpc CreateSubscription
 	var sub = &CreateSubscriptionData{
-		NetconfNotificationXmlns, "", "", "",
+		NetconfNotificationXmlns, "", "", "", "",
 	}
 	rpc.Subscription = *sub
 	rpc.MessageID = uuid()
 	return &rpc
 }
 
 // NewCreateSubscription can be used to create a `create-subscription` message.
-func NewCreateSubscription(stopTime string, startTime string, stream string) *CreateSubscription {
+func NewCreateSubscription(stopTime string, startTime string, stream string, filter string) *CreateSubscription {
 	var rpc CreateSubscription
 	var sub = &CreateSubscriptionData{
-		NetconfNotificationXmlns, stream, startTime, stopTime,
+		NetconfNotificationXmlns, stream, filter, startTime, stopTime,
 	}
 	rpc.Subscription = *sub
 	rpc.MessageID = uuid()

netconf/message/rpc.go

@@ -21,7 +21,8 @@ import (
 	"fmt"
 )
 
-const RpcReplyRegex = ".*rpc-reply"
+// RPCReplyRegex represents a regexp for RPC replies.
+const RPCReplyRegex = ".*rpc-reply"
 
 // RPCMethod defines the interface for creating an RPC method.
 // This is mostly a marker interface to abstract all RPCs