From cdb2f2d127f79a6d5ea2f9f2b4b7e4ac15fefb18 Mon Sep 17 00:00:00 2001 From: Kleissner Date: Wed, 29 Dec 2021 16:17:29 +0100 Subject: [PATCH] Update core library (refactoring). --- API.go | 124 ++++++++++++++++++++++++----------------------- Command Debug.go | 6 +-- Command Line.go | 48 +++++++++--------- File Transfer.go | 4 +- Main.go | 77 +++++++++++++++-------------- go.mod | 2 +- go.sum | 4 +- 7 files changed, 137 insertions(+), 128 deletions(-) diff --git a/API.go b/API.go index 21e11a6..1e2e5d6 100644 --- a/API.go +++ b/API.go @@ -39,11 +39,11 @@ func startAPI(backend *core.Backend, apiListen []string, apiKey uuid.UUID) { return } - api.InitGeoIPDatabase(config.GeoIPDatabase) + api.InitGeoIPDatabase(backend.Config.GeoIPDatabase) api.AllowKeyInParam = append(api.AllowKeyInParam, "/console") - api.Router.HandleFunc("/console", apiConsole).Methods("GET") - api.Router.HandleFunc("/shutdown", apiShutdown).Methods("GET") + api.Router.HandleFunc("/console", apiConsole(backend)).Methods("GET") + api.Router.HandleFunc("/shutdown", apiShutdown(backend)).Methods("GET") } // parseCmdParams parses the "-webapi", "-apikey", and "-watchpid" command line parameters. @@ -82,56 +82,58 @@ apiConsole provides a websocket to send/receive internal commands. Request: GET /console Result: Upgrade to websocket. The websocket message are texts to read/write. */ -func apiConsole(w http.ResponseWriter, r *http.Request) { - c, err := webapi.WSUpgrader.Upgrade(w, r, nil) - if err != nil { - // May happen if request is simple HTTP request. - return - } - defer c.Close() +func apiConsole(backend *core.Backend) func(w http.ResponseWriter, r *http.Request) { + return func(w http.ResponseWriter, r *http.Request) { + c, err := webapi.WSUpgrader.Upgrade(w, r, nil) + if err != nil { + // May happen if request is simple HTTP request. + return + } + defer c.Close() - bufferR := bytes.NewBuffer(make([]byte, 0, 4096)) - bufferW := bytes.NewBuffer(make([]byte, 0, 4096)) + bufferR := bytes.NewBuffer(make([]byte, 0, 4096)) + bufferW := bytes.NewBuffer(make([]byte, 0, 4096)) - terminateSignal := make(chan struct{}) - defer close(terminateSignal) + terminateSignal := make(chan struct{}) + defer close(terminateSignal) - // start userCommands which handles the actual commands - go userCommands(bufferR, bufferW, terminateSignal) + // start userCommands which handles the actual commands + go userCommands(backend, bufferR, bufferW, terminateSignal) - // go routine to receive output from userCommands and forward to websocket - go func() { - bufferW2 := make([]byte, 4096) + // go routine to receive output from userCommands and forward to websocket + go func() { + bufferW2 := make([]byte, 4096) + for { + select { + case <-terminateSignal: + return + default: + } + + countRead, err := bufferW.Read(bufferW2) + if err != nil || countRead == 0 { + time.Sleep(250 * time.Millisecond) + continue + } + + c.WriteMessage(websocket.TextMessage, bufferW2[:countRead]) + } + }() + + // read from websocket loop and forward to the userCommands routine for { - select { - case <-terminateSignal: - return - default: + _, message, err := c.ReadMessage() + if err != nil { // when channel is closed, an error is returned here + break } - countRead, err := bufferW.Read(bufferW2) - if err != nil || countRead == 0 { - time.Sleep(250 * time.Millisecond) - continue + // make sure the message has the \n delimiter which is used to detect a line + if !bytes.HasSuffix(message, []byte{'\n'}) { + message = append(message, '\n') } - c.WriteMessage(websocket.TextMessage, bufferW2[:countRead]) + bufferR.Write(message) } - }() - - // read from websocket loop and forward to the userCommands routine - for { - _, message, err := c.ReadMessage() - if err != nil { // when channel is closed, an error is returned here - break - } - - // make sure the message has the \n delimiter which is used to detect a line - if !bytes.HasSuffix(message, []byte{'\n'}) { - message = append(message, '\n') - } - - bufferR.Write(message) } } @@ -141,22 +143,24 @@ apiShutdown gracefully shuts down the application. Actions: 0 = Shutdown. Request: GET /shutdown?action=[action] Result: 200 with JSON structure apiShutdownStatus */ -func apiShutdown(w http.ResponseWriter, r *http.Request) { - r.ParseForm() - action, err := strconv.Atoi(r.Form.Get("action")) - if err != nil || action != 0 { - http.Error(w, "", http.StatusBadRequest) - return - } +func apiShutdown(backend *core.Backend) func(w http.ResponseWriter, r *http.Request) { + return func(w http.ResponseWriter, r *http.Request) { + r.ParseForm() + action, err := strconv.Atoi(r.Form.Get("action")) + if err != nil || action != 0 { + http.Error(w, "", http.StatusBadRequest) + return + } - if action == 0 { - // Later: Initiate shutdown signal to core library and wait for all requests to complete. + if action == 0 { + // Later: Initiate shutdown signal to core library and wait for all requests to complete. - core.Filters.LogError("apiShutdown", "graceful shutdown via API requested from '%s'\n", r.RemoteAddr) + backend.Filters.LogError("apiShutdown", "graceful shutdown via API requested from '%s'\n", r.RemoteAddr) - EncodeJSONFlush(w, r, &apiShutdownStatus{Status: 0}) + EncodeJSONFlush(backend, w, r, &apiShutdownStatus{Status: 0}) - os.Exit(core.ExitGraceful) + os.Exit(core.ExitGraceful) + } } } @@ -165,10 +169,10 @@ type apiShutdownStatus struct { } // EncodeJSONFlush encodes the data as JSON and flushes the writer. It sets the Content-Length header so no subsequent writes should be made. -func EncodeJSONFlush(w http.ResponseWriter, r *http.Request, data interface{}) (err error) { +func EncodeJSONFlush(backend *core.Backend, w http.ResponseWriter, r *http.Request, data interface{}) (err error) { response, err := json.Marshal(data) if err != nil { - core.Filters.LogError("EncodeJSONFlush", "error marshalling data for route '%s': %v\n", r.URL.Path, err) + backend.Filters.LogError("EncodeJSONFlush", "error marshalling data for route '%s': %v\n", r.URL.Path, err) return err } @@ -190,7 +194,7 @@ func EncodeJSONFlush(w http.ResponseWriter, r *http.Request, data interface{}) ( // It uses the command line parameter "-watchpid=[PID]". // This can be useful to automatically shut down the application in case the frontend shuts down unexpectedly. // Graceful shutdown should be initiated via the /shutdown API. -func processExitMonitor(watchPID int) { +func processExitMonitor(backend *core.Backend, watchPID int) { if watchPID == 0 { return } @@ -198,13 +202,13 @@ func processExitMonitor(watchPID int) { // monitor the process process, err := os.FindProcess(watchPID) if err != nil { - core.Filters.LogError("processExitMonitor", "error finding monitored process ID %d: %v\n", watchPID, err) + backend.Filters.LogError("processExitMonitor", "error finding monitored process ID %d: %v\n", watchPID, err) return } _, err = process.Wait() if err == nil { - core.Filters.LogError("processExitMonitor", "graceful shutdown via exit signal from process ID %d\n", watchPID) + backend.Filters.LogError("processExitMonitor", "graceful shutdown via exit signal from process ID %d\n", watchPID) // Later: Initiate shutdown signal to core library and wait for all requests to complete. diff --git a/Command Debug.go b/Command Debug.go index 90732f1..824c91f 100644 --- a/Command Debug.go +++ b/Command Debug.go @@ -20,12 +20,12 @@ import ( ) // debugCmdConnect connects to the node ID -func debugCmdConnect(nodeID []byte) { +func debugCmdConnect(backend *core.Backend, nodeID []byte) { fmt.Printf("---------------- Connect to node %s ----------------\n", hex.EncodeToString(nodeID)) defer fmt.Printf("---------------- done node %s ----------------\n", hex.EncodeToString(nodeID)) // in local DHT list? - _, peer := core.IsNodeContact(nodeID) + _, peer := backend.IsNodeContact(nodeID) if peer != nil { fmt.Printf("* In local routing table: Yes.\n") } else { @@ -35,7 +35,7 @@ func debugCmdConnect(nodeID []byte) { defer hashMonitorControl(nodeID, 1) // Discovery via DHT. - _, peer, _ = core.FindNode(nodeID, time.Second*10) + _, peer, _ = backend.FindNode(nodeID, time.Second*10) if peer == nil { fmt.Printf("* Not found via DHT :(\n") return diff --git a/Command Line.go b/Command Line.go index 70a2787..0eb6213 100644 --- a/Command Line.go +++ b/Command Line.go @@ -52,7 +52,7 @@ func showHelp(output io.Writer) { "\n") } -func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct{}) { +func userCommands(backend *core.Backend, input io.Reader, output io.Writer, terminateSignal chan struct{}) { reader := bufio.NewReader(input) fmt.Fprint(output, appName+" "+core.Version+"\n------------------------------\n") @@ -84,12 +84,12 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct fmt.Fprintf(output, "Public Key: %s\n", hex.EncodeToString(publicKey.SerializeCompressed())) case "debug key self": - privateKey, publicKey := core.ExportPrivateKey() + privateKey, publicKey := backend.ExportPrivateKey() fmt.Fprintf(output, "Private Key: %s\n", hex.EncodeToString(privateKey.Serialize())) fmt.Fprintf(output, "Public Key: %s\n", hex.EncodeToString(publicKey.SerializeCompressed())) case "peer list": - for _, peer := range GetPeerlistSorted() { + for _, peer := range GetPeerlistSorted(backend) { info := "" if peer.IsRootPeer { info = " [root peer]" @@ -104,18 +104,18 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct case "chat all", "chat": if text, valid, terminate := getUserOptionString(reader, terminateSignal); valid { - core.SendChatAll(text) + backend.SendChatAll(text) } else if terminate { return } case "status": - _, publicKey := core.ExportPrivateKey() - nodeID := core.SelfNodeID() + _, publicKey := backend.ExportPrivateKey() + nodeID := backend.SelfNodeID() fmt.Fprintf(output, "----------------\nPublic Key: %s\nNode ID: %s\n\n", hex.EncodeToString(publicKey.SerializeCompressed()), hex.EncodeToString(nodeID)) features := "" - featureSupport := core.FeatureSupport() + featureSupport := backend.FeatureSupport() if featureSupport&(1< 0 { features = "IPv4" } @@ -132,11 +132,11 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct features += "Firewall Reported" } - fmt.Fprintf(output, "User Agent: %s\nFeatures: %s\n\n", core.SelfUserAgent(), features) + fmt.Fprintf(output, "User Agent: %s\nFeatures: %s\n\n", backend.SelfUserAgent(), features) fmt.Fprintf(output, "Listen Address Multicast IP out External Address\n") - for _, network := range core.GetNetworks(4) { + for _, network := range backend.GetNetworks(4) { address, _, broadcastIPv4, ipExternal, externalPort := network.GetListen() broadcastIPsA := "" @@ -164,7 +164,7 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct fmt.Fprintf(output, "%-46s %-32s %s\n", address.String(), broadcastIPsA, externalAddress) } - for _, network := range core.GetNetworks(6) { + for _, network := range backend.GetNetworks(6) { address, multicastIP, _, _, externalPort := network.GetListen() externalPortA := "" @@ -176,7 +176,7 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct } fmt.Fprintf(output, "\nPeer ID Sent Received IP Flags RTT \n") - for _, peer := range GetPeerlistSorted() { + for _, peer := range GetPeerlistSorted(backend) { addressA := "N/A" rttA := "N/A" if connectionsActive := peer.GetConnections(true); len(connectionsActive) > 0 { @@ -210,7 +210,7 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct case "warehouse get": if hash, valid, terminate := getUserOptionHash(reader, terminateSignal); valid { - data, found := core.GetDataLocal(hash) + data, found := backend.GetDataLocal(hash) if !found { fmt.Fprintf(output, "Not found.\n") } else { @@ -225,7 +225,7 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct case "warehouse store": if text, valid, terminate := getUserOptionString(reader, terminateSignal); valid { - if err := core.StoreDataLocal([]byte(text)); err != nil { + if err := backend.StoreDataLocal([]byte(text)); err != nil { fmt.Fprintf(output, "Error storing data: %s\n", err.Error()) break } @@ -236,7 +236,7 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct case "dht store": if text, valid, terminate := getUserOptionString(reader, terminateSignal); valid { - if err := core.StoreDataDHT([]byte(text), 5); err != nil { + if err := backend.StoreDataDHT([]byte(text), 5); err != nil { fmt.Fprintf(output, "Error storing data: %s\n", err.Error()) break } @@ -247,7 +247,7 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct case "dht get": if hash, valid, terminate := getUserOptionHash(reader, terminateSignal); valid { - data, sender, found := core.GetDataDHT(hash) + data, sender, found := backend.GetDataDHT(hash) if !found { fmt.Fprintf(output, "Not found.\n") } else { @@ -309,12 +309,12 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct } // is self? - if bytes.Equal(nodeID, core.SelfNodeID()) { + if bytes.Equal(nodeID, backend.SelfNodeID()) { fmt.Fprintf(output, "Target node is self.\n") break } - debugCmdConnect(nodeID) + debugCmdConnect(backend, nodeID) case "debug watch searches": fmt.Fprintf(output, "Enable (1) or disable (0) watching of all outgoing DHT searches? (current setting: %t)\n", enableMonitorAll) @@ -394,9 +394,9 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct timeout := time.Second * 10 if valid2 { - peer, err = webapi.PeerConnectNode(nodeID, timeout) + peer, err = webapi.PeerConnectNode(backend, nodeID, timeout) } else if err3 == nil { - peer, err = webapi.PeerConnectPublicKey(publicKey, timeout) + peer, err = webapi.PeerConnectPublicKey(backend, publicKey, timeout) } if err != nil { fmt.Fprintf(output, "Could not connect to peer: %s\n", err.Error()) @@ -432,9 +432,9 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct timeout := time.Second * 10 if valid2 { - peer, err = webapi.PeerConnectNode(nodeID, timeout) + peer, err = webapi.PeerConnectNode(backend, nodeID, timeout) } else if err3 == nil { - peer, err = webapi.PeerConnectPublicKey(publicKey, timeout) + peer, err = webapi.PeerConnectPublicKey(backend, publicKey, timeout) } if err != nil { fmt.Fprintf(output, "Could not connect to peer: %s\n", err.Error()) @@ -444,7 +444,7 @@ func userCommands(input io.Reader, output io.Writer, terminateSignal chan struct go blockTransfer(peer, uint64(blockNumber)) case "exit": - core.Filters.LogError("userCommands", "graceful exit via user terminal command\n") + backend.Filters.LogError("userCommands", "graceful exit via user terminal command\n") os.Exit(core.ExitGraceful) case "search file": @@ -625,8 +625,8 @@ func connectionStatusToA(status int) (result string) { } } -func GetPeerlistSorted() (peers []*core.PeerInfo) { - peers = core.PeerlistGet() +func GetPeerlistSorted(backend *core.Backend) (peers []*core.PeerInfo) { + peers = backend.PeerlistGet() sort.Slice(peers, func(i, j int) bool { if peers[i].IsRootPeer && !peers[j].IsRootPeer { return true diff --git a/File Transfer.go b/File Transfer.go index 3e681b5..5c1dc9b 100644 --- a/File Transfer.go +++ b/File Transfer.go @@ -23,7 +23,7 @@ import ( // Note: The file MUST be stored locally, otherwise this function fails. func transferCompareFile(peer *core.PeerInfo, fileHash []byte) { // check if the file exists locally - _, fileInfo, status, _ := core.UserWarehouse.FileExists(fileHash) + _, fileInfo, status, _ := peer.Backend.UserWarehouse.FileExists(fileHash) if status != warehouse.StatusOK { fmt.Printf("File does not exist in local warehouse: %s\n", hex.EncodeToString(fileHash)) return @@ -99,7 +99,7 @@ func transferCompareFile(peer *core.PeerInfo, fileHash []byte) { // read the exact piece from the local file for comparison dataCompare := make([]byte, 0, n) compareBuffer := bytes.NewBuffer(dataCompare) - _, bytesRead, err := core.UserWarehouse.ReadFile(fileHash, int64(fileOffset), int64(n), compareBuffer) + _, bytesRead, err := peer.Backend.UserWarehouse.ReadFile(fileHash, int64(fileOffset), int64(n), compareBuffer) if err != nil { fmt.Printf("Warehouse error reading at offset %d length %d: %v\n", fileOffset, n, err) break diff --git a/Main.go b/Main.go index 9628312..98411c6 100644 --- a/Main.go +++ b/Main.go @@ -19,8 +19,7 @@ const appName = "Peernet Cmd" var config struct { // Log settings - ErrorOutput int `yaml:"ErrorOutput"` // 0 = Log file (default), 1 = Command line, 2 = Log file + command line, 3 = None - GeoIPDatabase string `yaml:"GeoIPDatabase"` // GeoLite2 City database to provide GeoIP information. + ErrorOutput int `yaml:"ErrorOutput"` // 0 = Log file (default), 1 = Command line, 2 = Log file + command line, 3 = None // API settings APIListen []string `yaml:"APIListen"` // WebListen is in format IP:Port and declares where the web-interface should listen on. IP can also be ommitted to listen on any. @@ -33,56 +32,62 @@ var config struct { } func init() { - if status, err := core.LoadConfigOut(configFile, &config); err != nil { - var exitCode int + if status, err := core.LoadConfig(configFile, &config); status != core.ExitSuccess { switch status { - case 0: - exitCode = core.ExitErrorConfigAccess + case core.ExitErrorConfigAccess: fmt.Printf("Unknown error accessing config file '%s': %s\n", configFile, err.Error()) - case 1: - exitCode = core.ExitErrorConfigRead + case core.ExitErrorConfigRead: fmt.Printf("Error reading config file '%s': %s\n", configFile, err.Error()) - case 2: - exitCode = core.ExitErrorConfigParse + case core.ExitErrorConfigParse: fmt.Printf("Error parsing config file '%s' (make sure it is valid YAML format): %s\n", configFile, err.Error()) default: - exitCode = core.ExitErrorConfigAccess fmt.Printf("Unknown error loading config file '%s': %s\n", configFile, err.Error()) } - os.Exit(exitCode) - } - - if err := core.InitLog(); err != nil { - fmt.Printf("Error opening log file: %s\n", err.Error()) - os.Exit(core.ExitErrorLogInit) + os.Exit(status) } monitorKeys = make(map[string]struct{}) - - core.Filters.LogError = logError - core.Filters.DHTSearchStatus = filterSearchStatus - core.Filters.IncomingRequest = filterIncomingRequest - core.Filters.MessageIn = filterMessageIn - core.Filters.MessageOutAnnouncement = filterMessageOutAnnouncement - core.Filters.MessageOutResponse = filterMessageOutResponse - core.Filters.MessageOutTraverse = filterMessageOutTraverse - core.Filters.MessageOutPing = filterMessageOutPing - core.Filters.MessageOutPong = filterMessageOutPong - - userAgent := appName + "/" + core.Version - - backend = core.Init(userAgent) } -var backend *core.Backend - func main() { + userAgent := appName + "/" + core.Version + + filters := &core.Filters{ + LogError: logError, + DHTSearchStatus: filterSearchStatus, + IncomingRequest: filterIncomingRequest, + MessageIn: filterMessageIn, + MessageOutAnnouncement: filterMessageOutAnnouncement, + MessageOutResponse: filterMessageOutResponse, + MessageOutTraverse: filterMessageOutTraverse, + MessageOutPing: filterMessageOutPing, + MessageOutPong: filterMessageOutPong, + } + + backend, status, err := core.Init(userAgent, configFile, filters) + + if status != core.ExitSuccess { + switch status { + case core.ExitErrorConfigAccess: + fmt.Printf("Unknown error accessing config file '%s': %s\n", configFile, err.Error()) + case core.ExitErrorConfigRead: + fmt.Printf("Error reading config file '%s': %s\n", configFile, err.Error()) + case core.ExitErrorConfigParse: + fmt.Printf("Error parsing config file '%s' (make sure it is valid YAML format): %s\n", configFile, err.Error()) + case core.ExitErrorLogInit: + fmt.Printf("Error opening log file '%s': %s\n", backend.Config.LogFile, err.Error()) + default: + fmt.Printf("Unknown error %d initializing backend: %s\n", status, err.Error()) + } + os.Exit(status) + } + apiListen, apiKey, watchPID := parseCmdParams() startAPI(backend, apiListen, apiKey) - go processExitMonitor(watchPID) + go processExitMonitor(backend, watchPID) - core.Connect() + backend.Connect() - userCommands(os.Stdin, os.Stdout, nil) + userCommands(backend, os.Stdin, os.Stdout, nil) } diff --git a/go.mod b/go.mod index b4d2299..05e0942 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/PeernetOfficial/Cmd go 1.16 require ( - github.com/PeernetOfficial/core v0.0.0-20211214164859-ade13d6422da + github.com/PeernetOfficial/core v0.0.0-20211229150740-0493acce6423 github.com/google/uuid v1.3.0 github.com/gorilla/websocket v1.4.3-0.20210424162022-e8629af678b7 ) diff --git a/go.sum b/go.sum index 168994d..8f74291 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ github.com/IncSW/geoip2 v0.1.1 h1:afzzYF7n9JbdcPy8aiBSgBJuXi4mTWXZ3z6V3o6Vg34= github.com/IncSW/geoip2 v0.1.1/go.mod h1:adcasR40vXiUBjtzdaTTKL/6wSf+fgO4M8Gve/XzPUk= -github.com/PeernetOfficial/core v0.0.0-20211214164859-ade13d6422da h1:mNK/4BcbCVJjjmWoUP839HUnIeP0GrBE8lxyldzOr8A= -github.com/PeernetOfficial/core v0.0.0-20211214164859-ade13d6422da/go.mod h1:VNhfwAZqya5JDZTS/ffa8Shof0OmQCx025CQMieGB7o= +github.com/PeernetOfficial/core v0.0.0-20211229150740-0493acce6423 h1:2y+D2hp/i63qMIWpmAT708VPYbDv4Dkx00xdCcXaXTY= +github.com/PeernetOfficial/core v0.0.0-20211229150740-0493acce6423/go.mod h1:VNhfwAZqya5JDZTS/ffa8Shof0OmQCx025CQMieGB7o= github.com/akrylysov/pogreb v0.10.1 h1:FqlR8VR7uCbJdfUob916tPM+idpKgeESDXOA1K0DK4w= github.com/akrylysov/pogreb v0.10.1/go.mod h1:pNs6QmpQ1UlTJKDezuRWmaqkgUE2TuU0YTWyqJZ7+lI= github.com/enfipy/locker v1.1.0 h1:2zVJ0ky7cS1Vjs0x6OQWFiT2dSEiHrI5/O2KCz1fgGc=