From 4cc764b7f05075c2a76936a556310d0796a2ea28 Mon Sep 17 00:00:00 2001 From: Ibrahim Ansari Date: Tue, 13 Feb 2024 01:16:54 +0530 Subject: [PATCH] Update xsync, rename managedProcess --- auth/authenticator.go | 2 +- auth/memory.go | 4 ++-- auth/redis.go | 2 +- auth/users.go | 4 ++-- connector.go | 18 +++++++++--------- endpoints_files.go | 6 +++--- endpoints_misc.go | 6 +++--- go.mod | 2 +- go.sum | 4 ++-- main.go | 2 +- process.go | 6 +++--- 11 files changed, 28 insertions(+), 28 deletions(-) diff --git a/auth/authenticator.go b/auth/authenticator.go index acaaf8a..1d8a3de 100644 --- a/auth/authenticator.go +++ b/auth/authenticator.go @@ -8,7 +8,7 @@ import ( "net/http" "sync" - "github.com/puzpuzpuz/xsync/v2" + "github.com/puzpuzpuz/xsync/v3" ) // Authenticator is used by Octyne's Connector to provide HTTP API authentication. diff --git a/auth/memory.go b/auth/memory.go index 2360104..c4c43c8 100644 --- a/auth/memory.go +++ b/auth/memory.go @@ -3,7 +3,7 @@ package auth import ( "net/http" - "github.com/puzpuzpuz/xsync/v2" + "github.com/puzpuzpuz/xsync/v3" ) // MemoryAuthenticator is an Authenticator implementation using an array to store tokens. @@ -15,7 +15,7 @@ type MemoryAuthenticator struct { // NewMemoryAuthenticator initializes an authenticator using memory for token storage. func NewMemoryAuthenticator(usersJsonPath string) Authenticator { users := CreateUserStore(usersJsonPath) - return &MemoryAuthenticator{Tokens: xsync.NewMapOf[string](), Users: users} + return &MemoryAuthenticator{Tokens: xsync.NewMapOf[string, string](), Users: users} } // GetUsers returns a Map with all the users and their corresponding passwords. diff --git a/auth/redis.go b/auth/redis.go index 44af028..668d00a 100644 --- a/auth/redis.go +++ b/auth/redis.go @@ -7,7 +7,7 @@ import ( "time" "github.com/gomodule/redigo/redis" - "github.com/puzpuzpuz/xsync/v2" + "github.com/puzpuzpuz/xsync/v3" ) // RedisAuthenticator is an Authenticator implementation using Redis to store tokens. diff --git a/auth/users.go b/auth/users.go index a0162e9..9ccd738 100644 --- a/auth/users.go +++ b/auth/users.go @@ -6,11 +6,11 @@ import ( "os" "time" - "github.com/puzpuzpuz/xsync/v2" + "github.com/puzpuzpuz/xsync/v3" ) func CreateUserStore(usersJsonPath string) *xsync.MapOf[string, string] { - var users = xsync.NewMapOf[string]() + var users = xsync.NewMapOf[string, string]() initialFile, updates, err := readAndWatchFile(usersJsonPath) if err != nil { log.Println("An error occurred while reading " + usersJsonPath + "! " + err.Error()) diff --git a/connector.go b/connector.go index b8ff2a3..d972e5e 100644 --- a/connector.go +++ b/connector.go @@ -13,7 +13,7 @@ import ( "github.com/gorilla/mux" "github.com/gorilla/websocket" - "github.com/puzpuzpuz/xsync/v2" + "github.com/puzpuzpuz/xsync/v3" "github.com/retrixe/octyne/auth" "go.uber.org/zap" "go.uber.org/zap/zapcore" @@ -51,8 +51,8 @@ func CreateZapLogger(config LoggingConfig) *zap.SugaredLogger { zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()), w, zap.InfoLevel)).Sugar() } -// An internal representation of Process along with the clients connected to it and its output. -type managedProcess struct { +// ExposedProcess contains Process along with connected clients and cached output. +type ExposedProcess struct { *Process Clients *xsync.MapOf[string, chan interface{}] Console string @@ -73,7 +73,7 @@ type Connector struct { *mux.Router *websocket.Upgrader *Logger - Processes *xsync.MapOf[string, *managedProcess] + Processes *xsync.MapOf[string, *ExposedProcess] Tickets *xsync.MapOf[string, Ticket] } @@ -110,8 +110,8 @@ func InitializeConnector(config *Config) *Connector { connector := &Connector{ Router: mux.NewRouter().StrictSlash(true), Logger: &Logger{LoggingConfig: config.Logging, Zap: CreateZapLogger(config.Logging)}, - Processes: xsync.NewMapOf[*managedProcess](), - Tickets: xsync.NewMapOf[Ticket](), + Processes: xsync.NewMapOf[string, *ExposedProcess](), + Tickets: xsync.NewMapOf[string, Ticket](), Authenticator: &auth.ReplaceableAuthenticator{Engine: authenticator}, Upgrader: &websocket.Upgrader{Subprotocols: []string{"console-v2"}}, } @@ -173,9 +173,9 @@ func InitializeConnector(config *Config) *Connector { // AddProcess adds a process to the connector to be accessed via the HTTP API. func (connector *Connector) AddProcess(proc *Process) { - process := &managedProcess{ + process := &ExposedProcess{ Process: proc, - Clients: xsync.NewMapOf[chan interface{}](), + Clients: xsync.NewMapOf[string, chan interface{}](), Console: "", } connector.Processes.Store(process.Name, process) @@ -263,7 +263,7 @@ func (connector *Connector) UpdateConfig(config *Config) { } } // Modify/remove existing processes. - connector.Processes.Range(func(key string, value *managedProcess) bool { + connector.Processes.Range(func(key string, value *ExposedProcess) bool { serverConfig, ok := config.Servers[key] if ok { value.Process.ServerConfigMutex.Lock() diff --git a/endpoints_files.go b/endpoints_files.go index 15ce7d2..2122589 100644 --- a/endpoints_files.go +++ b/endpoints_files.go @@ -19,7 +19,7 @@ import ( "time" "github.com/gorilla/mux" - "github.com/puzpuzpuz/xsync/v2" + "github.com/puzpuzpuz/xsync/v3" "github.com/retrixe/octyne/system" ) @@ -212,7 +212,7 @@ func fileEndpointPost(connector *Connector, w http.ResponseWriter, r *http.Reque } func fileEndpointPatch(connector *Connector, w http.ResponseWriter, r *http.Request, - process *managedProcess, id string, user string) { + process *ExposedProcess, id string, user string) { // Get the request body to check the operation. var body bytes.Buffer _, err := body.ReadFrom(r.Body) @@ -380,7 +380,7 @@ func folderEndpoint(connector *Connector, w http.ResponseWriter, r *http.Request // GET /server/{id}/compress?token=token // POST /server/{id}/compress?path=path&compress=true/false/zstd/xz/gzip&archiveType=zip/tar&basePath=path&async=boolean // POST /server/{id}/compress/v2?path=path&compress=true/false/zstd/xz/gzip&archiveType=zip/tar&basePath=path&async=boolean -var compressionProgressMap = xsync.NewMapOf[string]() +var compressionProgressMap = xsync.NewMapOf[string, string]() func compressionEndpoint(connector *Connector, w http.ResponseWriter, r *http.Request) { // Check with authenticator. diff --git a/endpoints_misc.go b/endpoints_misc.go index cd9d8b1..e746758 100644 --- a/endpoints_misc.go +++ b/endpoints_misc.go @@ -124,7 +124,7 @@ func serversEndpoint(connector *Connector, w http.ResponseWriter, r *http.Reques } // Get a map of processes and their online status. processes := make(map[string]interface{}) - connector.Processes.Range(func(k string, v *managedProcess) bool { + connector.Processes.Range(func(k string, v *ExposedProcess) bool { if r.URL.Query().Get("extrainfo") == "true" { processes[v.Name] = map[string]interface{}{ "status": v.Online.Load(), @@ -175,7 +175,7 @@ func serverEndpoint(connector *Connector, w http.ResponseWriter, r *http.Request } } -func serverEndpointGet(w http.ResponseWriter, process *managedProcess) { +func serverEndpointGet(w http.ResponseWriter, process *ExposedProcess) { // Get the PID of the process. var stat system.ProcessStats process.CommandMutex.RLock() @@ -211,7 +211,7 @@ func serverEndpointGet(w http.ResponseWriter, process *managedProcess) { } func serverEndpointPost(connector *Connector, w http.ResponseWriter, r *http.Request, - process *managedProcess, id string, user string) { + process *ExposedProcess, id string, user string) { // Get the request body to check whether the operation is to START or STOP. var body bytes.Buffer _, err := body.ReadFrom(r.Body) diff --git a/go.mod b/go.mod index 4060a64..1fdd33f 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/gorilla/handlers v1.4.2 github.com/gorilla/mux v1.7.4 github.com/gorilla/websocket v1.4.2 - github.com/puzpuzpuz/xsync/v2 v2.4.0 + github.com/puzpuzpuz/xsync/v3 v3.0.2 github.com/tailscale/hujson v0.0.0-20221223112325-20486734a56a go.uber.org/zap v1.24.0 gopkg.in/natefinch/lumberjack.v2 v2.2.1 diff --git a/go.sum b/go.sum index 20594a8..1670cfc 100644 --- a/go.sum +++ b/go.sum @@ -14,8 +14,8 @@ github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/ad github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/puzpuzpuz/xsync/v2 v2.4.0 h1:5sXAMHrtx1bg9nbRZTOn8T4MkWe5V+o8yKRH02Eznag= -github.com/puzpuzpuz/xsync/v2 v2.4.0/go.mod h1:gD2H2krq/w52MfPLE+Uy64TzJDVY7lP2znR9qmR35kU= +github.com/puzpuzpuz/xsync/v3 v3.0.2 h1:3yESHrRFYr6xzkz61LLkvNiPFXxJEAABanTQpKbAaew= +github.com/puzpuzpuz/xsync/v3 v3.0.2/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk= diff --git a/main.go b/main.go index 07874b7..5cb6519 100644 --- a/main.go +++ b/main.go @@ -83,7 +83,7 @@ func main() { } else if err := connector.Logger.Zap.Sync(); err != nil { log.Println("Error when syncing the logger!", err) } - // TODO: connector.Processes.Range(func(key string, value *managedProcess) bool { value.StopProcess() }) + // TODO: connector.Processes.Range(func(key string, value *ExposedProcess) bool { value.StopProcess() }) os.Exit(exitCode) })() diff --git a/process.go b/process.go index fc7de08..8206bcc 100644 --- a/process.go +++ b/process.go @@ -147,13 +147,13 @@ func (process *Process) MonitorProcess(connector *Connector) error { if process.ToDelete.Load() { process.SendConsoleOutput("[Octyne] Server " + process.Name + " was marked for deletion, " + "stopped/crashed, and has now been removed.") - if managedProcess, loaded := connector.Processes.LoadAndDelete(process.Name); loaded { + if process, loaded := connector.Processes.LoadAndDelete(process.Name); loaded { <-time.After(5 * time.Second) - managedProcess.Clients.Range(func(key string, ws chan interface{}) bool { + process.Clients.Range(func(key string, ws chan interface{}) bool { ws <- nil return true }) - managedProcess.Clients.Clear() + process.Clients.Clear() } } else if process.Command.ProcessState.Success() || process.Online.Load() == 0 /* SIGKILL (if done by Octyne) */ ||