diff --git a/services/audit/pkg/command/server.go b/services/audit/pkg/command/server.go
index f28a9e3fb..ac10a5b68 100644
--- a/services/audit/pkg/command/server.go
+++ b/services/audit/pkg/command/server.go
@@ -8,7 +8,7 @@ import (
 	"os"
 
 	"github.com/cs3org/reva/v2/pkg/events"
-	"github.com/cs3org/reva/v2/pkg/events/server"
+	"github.com/cs3org/reva/v2/pkg/events/stream"
 	"github.com/go-micro/plugins/v4/events/natsjs"
 	"github.com/owncloud/ocis/v2/ocis-pkg/config/configlog"
 	ociscrypto "github.com/owncloud/ocis/v2/ocis-pkg/crypto"
@@ -63,7 +63,7 @@ func Server(cfg *config.Config) *cli.Command {
 					RootCAs:            rootCAPool,
 				}
 			}
-			client, err := server.NewNatsStream(
+			client, err := stream.Nats(
 				natsjs.TLSConfig(tlsConf),
 				natsjs.Address(evtsCfg.Endpoint),
 				natsjs.ClusterID(evtsCfg.Cluster),
diff --git a/services/graph/pkg/server/http/server.go b/services/graph/pkg/server/http/server.go
index 8951c5853..89070cd11 100644
--- a/services/graph/pkg/server/http/server.go
+++ b/services/graph/pkg/server/http/server.go
@@ -7,7 +7,7 @@ import (
 	stdhttp "net/http"
 	"os"
 
-	"github.com/cs3org/reva/v2/pkg/events/server"
+	"github.com/cs3org/reva/v2/pkg/events/stream"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	chimiddleware "github.com/go-chi/chi/v5/middleware"
 	"github.com/go-micro/plugins/v4/events/natsjs"
@@ -74,7 +74,7 @@ func Server(opts ...Option) (http.Service, error) {
 				RootCAs:            rootCAPool,
 			}
 		}
-		publisher, err = server.NewNatsStream(
+		publisher, err = stream.Nats(
 			natsjs.TLSConfig(tlsConf),
 			natsjs.Address(options.Config.Events.Endpoint),
 			natsjs.ClusterID(options.Config.Events.Cluster),
diff --git a/services/notifications/pkg/command/server.go b/services/notifications/pkg/command/server.go
index bef9c13c9..9c82a2d7d 100644
--- a/services/notifications/pkg/command/server.go
+++ b/services/notifications/pkg/command/server.go
@@ -7,7 +7,7 @@ import (
 	"os"
 
 	"github.com/cs3org/reva/v2/pkg/events"
-	"github.com/cs3org/reva/v2/pkg/events/server"
+	"github.com/cs3org/reva/v2/pkg/events/stream"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/go-micro/plugins/v4/events/natsjs"
 	"github.com/owncloud/ocis/v2/ocis-pkg/config/configlog"
@@ -62,7 +62,7 @@ func Server(cfg *config.Config) *cli.Command {
 					RootCAs:            rootCAPool,
 				}
 			}
-			client, err := server.NewNatsStream(
+			client, err := stream.Nats(
 				natsjs.TLSConfig(tlsConf),
 				natsjs.Address(evtsCfg.Endpoint),
 				natsjs.ClusterID(evtsCfg.Cluster),
diff --git a/services/postprocessing/pkg/command/server.go b/services/postprocessing/pkg/command/server.go
index 09540f4ab..be2010d15 100644
--- a/services/postprocessing/pkg/command/server.go
+++ b/services/postprocessing/pkg/command/server.go
@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/cs3org/reva/v2/pkg/events/server"
+	"github.com/cs3org/reva/v2/pkg/events/stream"
 	"github.com/go-micro/plugins/v4/events/natsjs"
 	ociscrypto "github.com/owncloud/ocis/v2/ocis-pkg/crypto"
 	"github.com/owncloud/ocis/v2/services/postprocessing/pkg/config"
@@ -56,7 +56,7 @@ func Server(cfg *config.Config) *cli.Command {
 				}
 			}
 
-			bus, err := server.NewNatsStream(
+			bus, err := stream.Nats(
 				natsjs.TLSConfig(tlsConf),
 				natsjs.Address(evtsCfg.Endpoint),
 				natsjs.ClusterID(evtsCfg.Cluster),
diff --git a/services/search/pkg/service/grpc/v0/service.go b/services/search/pkg/service/grpc/v0/service.go
index 712896f54..925eeb6fb 100644
--- a/services/search/pkg/service/grpc/v0/service.go
+++ b/services/search/pkg/service/grpc/v0/service.go
@@ -14,7 +14,7 @@ import (
 	ctxpkg "github.com/cs3org/reva/v2/pkg/ctx"
 	revactx "github.com/cs3org/reva/v2/pkg/ctx"
 	"github.com/cs3org/reva/v2/pkg/errtypes"
-	"github.com/cs3org/reva/v2/pkg/events/server"
+	"github.com/cs3org/reva/v2/pkg/events/stream"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/go-micro/plugins/v4/events/natsjs"
 	"github.com/jellydator/ttlcache/v2"
@@ -98,7 +98,7 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 			RootCAs:            rootCAPool,
 		}
 	}
-	bus, err := server.NewNatsStream(
+	bus, err := stream.Nats(
 		natsjs.TLSConfig(tlsConf),
 		natsjs.Address(cfg.Events.Endpoint),
 		natsjs.ClusterID(cfg.Events.Cluster),