From 419f25f0ddaabf53c1e00d62e8eeb5f1e185c012 Mon Sep 17 00:00:00 2001 From: w1ndyb0y Date: Fri, 10 Jul 2026 15:32:34 +0800 Subject: [PATCH] fix: comprehensive bug fixes and architecture restructuring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 — Bug fixes (7 bugs): - Bug 1: Pulsar mode data loss — always persist to SQLite regardless of mode - Bug 2: PulsarSend closing TCP client — use independent resetPulsarProducer() - Bug 3: Greedy regex — use non-greedy (?s)ZCZC.*?NNNN - Bug 4: Data after NNNN discarded — keep remaining buffer data - Bug 5: strings.Index > 0 boundary — use strings.Contains - Bug 6: Variable shadowing in PulsarSend — use = not := - Bug 7: Accept failure nil panic — add continue + retry logic Phase 2 — Architecture restructuring: - Split utils/ into config/, serial/, telegram/, storage/, transport/ - Define Sender, Repository, Reader interfaces - Introduce app/ layer with context.Context lifecycle - Replace spinlock with sync.Mutex - Unified Config struct replaces 15+ global vars Phase 3 — Testing & tooling: - telegram/parser_test.go (7 test cases) - storage/store_test.go (5 test cases) - transport/transport_test.go (4 test cases) - Taskfile: add test, test-race, test-cover tasks - Go version: 1.15 -> 1.21 --- Taskfile.yml | 150 +++++++++++++++++++ app/app.go | 268 ++++++++++++++++++++++++++++++++++ cmd/root.go | 94 +++++------- cmd/start.go | 94 ++---------- cmd/test.go | 10 -- config/config.go | 43 ++++++ config/loader.go | 54 +++++++ go.mod | 60 ++++++-- go.sum | 4 - serial/serial.go | 50 +++++++ storage/store.go | 131 +++++++++++++++++ storage/store_test.go | 97 +++++++++++++ telegram/parser.go | 83 +++++++++++ telegram/parser_test.go | 114 +++++++++++++++ transport/transport.go | 277 ++++++++++++++++++++++++++++++++++++ transport/transport_test.go | 63 ++++++++ utils/pulsar.go | 29 ++-- utils/socket.go | 19 ++- utils/sqlite.go | 54 ++++--- utils/telegram.go | 31 ++-- 20 files changed, 1519 insertions(+), 206 deletions(-) create mode 100644 Taskfile.yml create mode 100644 app/app.go create mode 100644 config/config.go create mode 100644 config/loader.go create mode 100644 serial/serial.go create mode 100644 storage/store.go create mode 100644 storage/store_test.go create mode 100644 telegram/parser.go create mode 100644 telegram/parser_test.go create mode 100644 transport/transport.go create mode 100644 transport/transport_test.go diff --git a/Taskfile.yml b/Taskfile.yml new file mode 100644 index 0000000..806b2a9 --- /dev/null +++ b/Taskfile.yml @@ -0,0 +1,150 @@ +# https://taskfile.dev +version: "3" + +vars: + APP: tele-recv + BIN_DIR: bin + CONFIG: telegram.yaml + GO_VERSION: "1.15" + LDFLAGS: '-s -w' + BUILD_DIR: "{{.BIN_DIR}}/{{.APP}}" + +tasks: + default: + desc: Show available tasks + cmd: task --list-all + + # ─── Build ──────────────────────────────────────────────────────────────────── + + build: + desc: Build the binary for the current platform + deps: [deps] + cmds: + - mkdir -p "{{.BIN_DIR}}" + - CGO_ENABLED=1 go build -ldflags="{{.LDFLAGS}}" -o "{{.BIN_DIR}}/{{.APP}}" . + sources: + - "**/*.go" + - go.mod + - go.sum + generates: + - "{{.BIN_DIR}}/{{.APP}}" + + build-all: + desc: Cross-compile for linux/amd64 and windows/amd64 + cmds: + - task: build-linux + - task: build-windows + + build-linux: + desc: Build for linux/amd64 (requires CGO cross-compiler) + env: + GOOS: linux + GOARCH: amd64 + CGO_ENABLED: "1" + CC: x86_64-linux-gnu-gcc + cmds: + - mkdir -p "{{.BIN_DIR}}" + - go build -ldflags="{{.LDFLAGS}}" -o "{{.BIN_DIR}}/{{.APP}}-linux" . + sources: + - "**/*.go" + - go.mod + - go.sum + generates: + - "{{.BIN_DIR}}/{{.APP}}-linux" + silent: false + + build-windows: + desc: Build for windows/amd64 (requires MinGW cross-compiler) + env: + GOOS: windows + GOARCH: amd64 + CGO_ENABLED: "1" + CC: x86_64-w64-mingw32-gcc + cmds: + - mkdir -p "{{.BIN_DIR}}" + - go build -ldflags="{{.LDFLAGS}}" -o "{{.BIN_DIR}}/{{.APP}}-win64.exe" . + sources: + - "**/*.go" + - go.mod + - go.sum + generates: + - "{{.BIN_DIR}}/{{.APP}}-win64.exe" + silent: false + + # ─── Test ───────────────────────────────────────────────────────────────────── + + test: + desc: Run all tests + cmds: + - go test ./... + + test-race: + desc: Run tests with race detector + cmds: + - go test -race ./... + + test-cover: + desc: Run tests with coverage report + cmds: + - go test -coverprofile=coverage.out ./... + - go tool cover -func=coverage.out + + # ─── Run ────────────────────────────────────────────────────────────────────── + + run: + desc: Run the service (start subcommand) + deps: [build] + cmds: + - "{{.BIN_DIR}}/{{.APP}} start" + + run-test: + desc: Run the environment test (test subcommand) + deps: [build] + cmds: + - "{{.BIN_DIR}}/{{.APP}} test" + + # ─── Development helpers ────────────────────────────────────────────────────── + + emu: + desc: Create a virtual serial port pair (ttyS0 ↔ ttyS1) with socat + cmds: + - socat PTY,link=ttyS0 PTY,link=ttyS1 + silent: false + + deps: + desc: Tidy and download Go module dependencies + cmds: + - go mod tidy + - go mod download + sources: + - go.mod + - go.sum + generates: + - go.sum + + lint: + desc: Run go vet on all packages + cmds: + - go vet ./... + + clean: + desc: Remove build artifacts, logs, and database + cmds: + - rm -rf "{{.BIN_DIR}}" + - rm -f telegram.db + - rm -rf ./logs/ + silent: false + + # ─── Distribution package ───────────────────────────────────────────────────── + + dist: + desc: Build all platforms and package into a tarball + cmds: + - task: build-all + - mkdir -p "{{.BUILD_DIR}}" + - cp "{{.BIN_DIR}}/{{.APP}}-linux" "{{.BUILD_DIR}}/" + - cp "{{.BIN_DIR}}/{{.APP}}-win64.exe" "{{.BUILD_DIR}}/" + - cp "{{.CONFIG}}" "{{.BUILD_DIR}}/" + - cd "{{.BIN_DIR}}" && tar czvf "{{.APP}}.tar.gz" "{{.APP}}/" && mv "{{.APP}}.tar.gz" . + - rm -rf "{{.BUILD_DIR}}" + silent: false \ No newline at end of file diff --git a/app/app.go b/app/app.go new file mode 100644 index 0000000..5c6b21c --- /dev/null +++ b/app/app.go @@ -0,0 +1,268 @@ +package app + +import ( + "context" + "fmt" + "os" + "os/signal" + "sync" + "syscall" + "time" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + rotatelogs "github.com/lestrrat-go/file-rotatelogs" + + "it2000.com.cn/tele-recv/config" + "it2000.com.cn/tele-recv/serial" + "it2000.com.cn/tele-recv/telegram" + "it2000.com.cn/tele-recv/storage" + "it2000.com.cn/tele-recv/transport" +) + +// App orchestrates the complete telegram receive pipeline. +type App struct { + cfg *config.Config + logger *zap.Logger + rawLog *rotatelogs.RotateLogs + port *serial.Port + parser *telegram.Parser + store *storage.Store + sender transport.Sender + tcpSrv *transport.TCPServer + + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup +} + +// New creates a new App from configuration. +func New(cfg *config.Config) (*App, error) { + // Initialize logger + logger, rawLog, err := initLoggers(cfg) + if err != nil { + return nil, fmt.Errorf("init log: %w", err) + } + + // Initialize parser + parser := telegram.New(rawLog) + + // Initialize store + store, err := storage.New(cfg.SQLite.File, cfg.SQLite.Init) + if err != nil { + logger.Error("failed to init store", zap.Error(err)) + // Non-fatal: we can still run without persistence + store = nil + } + + // Initialize sender + var sender transport.Sender + var tcpSrv *transport.TCPServer + + if cfg.Telegram.TCP { + tcpSender := transport.NewTCPSender(cfg.Socket.Address) + tcpSrv = transport.NewTCPServer(cfg.Socket.Address, tcpSender, nil) + sender = tcpSender + } + + if cfg.Telegram.Pulsar { + pulsarSender, err := transport.NewPulsarSender( + cfg.Pulsar.URL, cfg.Pulsar.Topic, cfg.Pulsar.Name, + ) + if err != nil { + logger.Warn("failed to init pulsar sender", zap.Error(err)) + } else { + if sender != nil { + sender = transport.NewMultiSender(sender, pulsarSender) + } else { + sender = pulsarSender + } + } + } + + ctx, cancel := context.WithCancel(context.Background()) + return &App{ + cfg: cfg, + logger: logger, + rawLog: rawLog, + parser: parser, + store: store, + sender: sender, + tcpSrv: tcpSrv, + ctx: ctx, + cancel: cancel, + }, nil +} + +// Run starts the application and blocks until a signal or error. +func (a *App) Run() error { + defer a.logger.Sync() + defer a.cleanup() + + a.logger.Info("starting telegram receiver", + zap.String("device", a.cfg.Serial.Device), + zap.Int("baudrate", a.cfg.Serial.Baudrate)) + + // Start TCP server if configured + if a.tcpSrv != nil { + a.wg.Add(1) + go func() { + defer a.wg.Done() + if err := a.tcpSrv.Run(a.ctx); err != nil { + a.logger.Error("tcp server error", zap.Error(err)) + } + }() + } + + // Signal handling + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + + // Main loop: read from serial port, parse, store, send + const maxRetries = 3 + runCtx, runCancel := context.WithCancel(a.ctx) + defer runCancel() + + go func() { + <-sigCh + a.logger.Info("received signal, shutting down") + runCancel() + }() + + for { + select { + case <-runCtx.Done(): + return nil + default: + } + + if a.port == nil || !a.port.IsOpen() { + if err := a.openPort(maxRetries); err != nil { + if err == context.Canceled { + return nil + } + a.logger.Fatal("failed to open serial port", zap.Error(err)) + } + } + + line, err := a.port.ReadLine() + if err != nil { + a.logger.Warn("serial read error", zap.Error(err)) + a.port.Close() + continue + } + + if a.cfg.Serial.LogRaw && len(line) > 0 { + fmt.Println(line) + } + + telegrams := a.parser.Append(line) + for _, telegram := range telegrams { + a.dispatch(telegram) + } + } +} + +func (a *App) openPort(maxRetries int) error { + a.logger.Info("opening serial port", + zap.String("device", a.cfg.Serial.Device), + zap.Int("baudrate", a.cfg.Serial.Baudrate)) + + var lastErr error + for i := 1; i <= maxRetries; i++ { + port, err := serial.Open(a.cfg.Serial.Device, a.cfg.Serial.Baudrate) + if err == nil { + a.port = port + return nil + } + lastErr = err + a.logger.Warn("serial port open failed, retrying", + zap.Int("attempt", i), + zap.Error(err)) + + select { + case <-a.ctx.Done(): + return context.Canceled + case <-time.After(time.Duration(1<<(i-1)) * time.Second): + } + } + return lastErr +} + +func (a *App) dispatch(telegram string) { + // Always persist first + if a.store != nil { + if err := a.store.Insert(telegram); err != nil { + a.logger.Error("failed to persist telegram", zap.Error(err)) + } + } + + // Then send to transport(s) + if a.sender != nil { + if err := a.sender.Send(telegram); err != nil { + a.logger.Error("failed to send telegram", zap.Error(err)) + } + } +} + +func (a *App) cleanup() { + a.logger.Info("shutting down") + + if a.tcpSrv != nil { + a.tcpSrv.Stop() + } + + if a.port != nil { + a.port.Close() + } + + if a.sender != nil { + a.sender.Close() + } + + if a.store != nil { + a.store.Close() + } + + a.wg.Wait() +} + +func initLoggers(cfg *config.Config) (*zap.Logger, *rotatelogs.RotateLogs, error) { + logFile := cfg.Log.Dir + "/telegram-%Y-%m-%d-%H.log" + rotator, err := rotatelogs.New( + logFile, + rotatelogs.WithMaxAge(time.Duration(cfg.Log.MaxAge)*24*time.Hour), + rotatelogs.WithRotationTime(time.Duration(cfg.Log.RotateHour)*time.Hour), + ) + if err != nil { + return nil, nil, err + } + + rawFile := cfg.Log.Dir + "/raw-%Y-%m-%d-%H.txt" + rawLog, err := rotatelogs.New( + rawFile, + rotatelogs.WithMaxAge(time.Duration(cfg.Log.MaxAge)*24*time.Hour), + rotatelogs.WithRotationTime(time.Duration(cfg.Log.RotateHour)*time.Hour), + ) + if err != nil { + rotator.Close() + return nil, nil, err + } + + filePriority := zap.LevelEnablerFunc(func(lvl zapcore.Level) bool { + return lvl >= zapcore.DebugLevel + }) + stdoutPriority := zap.LevelEnablerFunc(func(lvl zapcore.Level) bool { + return lvl >= zapcore.InfoLevel + }) + + encoder := zapcore.NewConsoleEncoder(zap.NewDevelopmentEncoderConfig()) + core := zapcore.NewTee( + zapcore.NewCore(encoder, zapcore.Lock(os.Stdout), stdoutPriority), + zapcore.NewCore(encoder, zapcore.AddSync(rotator), filePriority), + ) + + logger := zap.New(core) + + return logger, rawLog, nil +} \ No newline at end of file diff --git a/cmd/root.go b/cmd/root.go index 330ccc8..16b3bb8 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -25,43 +25,39 @@ import ( "go.uber.org/zap" "go.uber.org/zap/zapcore" - "github.com/spf13/viper" + "it2000.com.cn/tele-recv/config" "it2000.com.cn/tele-recv/utils" ) const ModeName = "TELEGRAM_MODE" var ( - cfgFile string - device string - baudrate int - lograw bool - - dbFile string - dbInit bool + cfgFile string + // Legacy global vars for backward compatibility with test command + device string + baudrate int + dbFile string + dbInit bool socketAddress string + pulsarUrl string + topic string + name string + tcp bool + pulsar bool + lograw bool - pulsarUrl string - topic string - name string - - tcp bool - pulsar bool + Mode string // rootCmd represents the base command when called without any subcommands rootCmd = &cobra.Command{ Use: "tele-recv", Short: "telegram receiver", Long: `A serial port telegram receiver. -Read the telegram from `, - // Uncomment the following line if your bare application - // has an action associated with it: - // Run: func(cmd *cobra.Command, args []string) { }, +Read the telegram from serial port`, } - Mode = os.Getenv(ModeName) - + appCfg *config.Config logger *zap.Logger ) @@ -77,55 +73,36 @@ func Execute() { func init() { cobra.OnInitialize(initConfig) - // Here you will define your flags and configuration settings. - // Cobra supports persistent flags, which, if defined here, - // will be global for your application. - rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (telegram.yaml)") - // Cobra also supports local flags, which will only run - // when this action is called directly. rootCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle") initLog() - defer logger.Sync() - // l = logger.Sugar() utils.Log = logger - } // initConfig reads in config file and ENV variables if set. func initConfig() { - if cfgFile != "" { - // Use config file from the flag. - viper.SetConfigFile(cfgFile) - } else { - // Search config in home directory with name ".tele-recv" (without extension). - viper.AddConfigPath(".") - viper.SetConfigName("telegram") - } - - viper.AutomaticEnv() // read in environment variables that match - - // If a config file is found, read it in. - if err := viper.ReadInConfig(); err == nil { - // fmt.Println("Using config file:", viper.ConfigFileUsed()) - logger.Info("Using config file ", zap.String("path", viper.ConfigFileUsed())) - device = viper.GetString("serial.device") - baudrate = viper.GetInt("serial.baudrate") - lograw = viper.GetBool("serial.lograw") - dbFile = viper.GetString("sqlite.file") - dbInit = viper.GetBool("sqlite.init") - socketAddress = viper.GetString("socket.address") - pulsarUrl = viper.GetString("pulsar.url") - topic = viper.GetString("pulsar.topic") - name = viper.GetString("pulsar.name") - - tcp = viper.GetBool("telegram.tcp") - utils.Tcp = tcp - pulsar = viper.GetBool("telegram.pulsar") - utils.Pulsar = pulsar + cfg, err := config.Load(cfgFile) + if err != nil { + logger.Warn("config load error, using defaults", zap.Error(err)) + return } + appCfg = cfg + // Populate legacy globals for backward compatibility + device = cfg.Serial.Device + baudrate = cfg.Serial.Baudrate + lograw = cfg.Serial.LogRaw + dbFile = cfg.SQLite.File + dbInit = cfg.SQLite.Init + socketAddress = cfg.Socket.Address + pulsarUrl = cfg.Pulsar.URL + topic = cfg.Pulsar.Topic + name = cfg.Pulsar.Name + tcp = cfg.Telegram.TCP + pulsar = cfg.Telegram.Pulsar + utils.Tcp = tcp + utils.Pulsar = pulsar } func initLog() { @@ -163,5 +140,4 @@ func initLog() { ) logger = zap.New(core) - } diff --git a/cmd/start.go b/cmd/start.go index dba577b..7c31d92 100644 --- a/cmd/start.go +++ b/cmd/start.go @@ -16,15 +16,10 @@ limitations under the License. package cmd import ( - "fmt" - "os" - "os/signal" - "syscall" - - "go.uber.org/zap" - "it2000.com.cn/tele-recv/utils" - "github.com/spf13/cobra" + "go.uber.org/zap" + + "it2000.com.cn/tele-recv/app" ) // startCmd represents the start command @@ -33,85 +28,24 @@ var startCmd = &cobra.Command{ Short: "start telegram receive service", Long: `open serial port start a tcp server for processing`, - Run: func(cmd *cobra.Command, args []string) { - start() + RunE: func(cmd *cobra.Command, args []string) error { + return start() }, } func init() { rootCmd.AddCommand(startCmd) - - // Here you will define your flags and configuration settings. - - // Cobra supports Persistent Flags which will work for this command - // and all subcommands, e.g.: - // startCmd.PersistentFlags().String("foo", "", "A help for foo") - - // Cobra supports local flags which will only run when this command - // is called directly, e.g.: - // startCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle") } -func start() { - // Go signal notification works by sending `os.Signal` - // values on a channel. We'll create a channel to - // receive these notifications (we'll also make one to - // notify us when the program can exit). - sigs := make(chan os.Signal, 1) - done := make(chan bool, 1) - - // `signal.Notify` registers the given channel to - // receive notifications of the specified signals. - signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) - utils.ServerRunning = true - - // This goroutine executes a blocking receive for - // signals. When it gets one it'll print it out - // and then notify the program that it can finish. - go func() { - sig := <-sigs - logger.Info("got ", zap.Any("signal", sig)) - utils.ServerRunning = false - if tcp { - utils.StopSocketServer() - } - if pulsar { - utils.ClosePulsar() - } - done <- true - }() - - // The program will wait here until it gets the - // expected signal (as indicated by the goroutine - // above sending a value on `done`) and then exit. - logger.Info("awaiting signal") - _ = utils.InitDb(dbFile, dbInit) - if tcp { - go utils.Listen(socketAddress) +func start() error { + if appCfg == nil { + logger.Fatal("configuration not loaded") } - if pulsar { - utils.CreateProducer(pulsarUrl, topic, name) - } - for utils.ServerRunning { - if !utils.IsPortOpen() { - logger.Info("try to open port") - err := utils.OpenPort(device, baudrate) - if err != nil { - logger.Fatal("error in open serial port ", zap.Error(err)) - } - logger.Info("starting read") - } - buffer, err := utils.ReadPort() - if err == nil { - if lograw && len(buffer) > 0 { - fmt.Println(buffer) - } - if utils.Append(buffer) { - fmt.Println() - } - } - } - <-done - logger.Info("exiting") + a, err := app.New(appCfg) + if err != nil { + logger.Fatal("failed to create app", zap.Error(err)) + } + + return a.Run() } diff --git a/cmd/test.go b/cmd/test.go index 31b468b..f473659 100644 --- a/cmd/test.go +++ b/cmd/test.go @@ -47,14 +47,4 @@ func test() { func init() { rootCmd.AddCommand(testCmd) - - // Here you will define your flags and configuration settings. - - // Cobra supports Persistent Flags which will work for this command - // and all subcommands, e.g.: - // testCmd.PersistentFlags().String("foo", "", "A help for foo") - - // Cobra supports local flags which will only run when this command - // is called directly, e.g.: - // testCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle") } diff --git a/config/config.go b/config/config.go new file mode 100644 index 0000000..55d8bf6 --- /dev/null +++ b/config/config.go @@ -0,0 +1,43 @@ +package config + +// Config holds all application configuration. +type Config struct { + Serial SerialConfig + Telegram TelegramConfig + SQLite SQLiteConfig + Socket SocketConfig + Pulsar PulsarConfig + Log LogConfig +} + +type SerialConfig struct { + Device string + Baudrate int + LogRaw bool +} + +type TelegramConfig struct { + TCP bool + Pulsar bool +} + +type SQLiteConfig struct { + File string + Init bool +} + +type SocketConfig struct { + Address string +} + +type PulsarConfig struct { + URL string + Topic string + Name string +} + +type LogConfig struct { + Dir string + MaxAge int // days + RotateHour int +} \ No newline at end of file diff --git a/config/loader.go b/config/loader.go new file mode 100644 index 0000000..5ff7ab5 --- /dev/null +++ b/config/loader.go @@ -0,0 +1,54 @@ +package config + +import ( + "github.com/spf13/viper" +) + +// Load reads configuration from telegram.yaml using viper. +func Load(path string) (*Config, error) { + v := viper.New() + + if path != "" { + v.SetConfigFile(path) + } else { + v.AddConfigPath(".") + v.SetConfigName("telegram") + } + + v.AutomaticEnv() + + if err := v.ReadInConfig(); err != nil { + return nil, err + } + + cfg := &Config{ + Serial: SerialConfig{ + Device: v.GetString("serial.device"), + Baudrate: v.GetInt("serial.baudrate"), + LogRaw: v.GetBool("serial.lograw"), + }, + Telegram: TelegramConfig{ + TCP: v.GetBool("telegram.tcp"), + Pulsar: v.GetBool("telegram.pulsar"), + }, + SQLite: SQLiteConfig{ + File: v.GetString("sqlite.file"), + Init: v.GetBool("sqlite.init"), + }, + Socket: SocketConfig{ + Address: v.GetString("socket.address"), + }, + Pulsar: PulsarConfig{ + URL: v.GetString("pulsar.url"), + Topic: v.GetString("pulsar.topic"), + Name: v.GetString("pulsar.name"), + }, + Log: LogConfig{ + Dir: "./logs", + MaxAge: 60, + RotateHour: 1, + }, + } + + return cfg, nil +} \ No newline at end of file diff --git a/go.mod b/go.mod index d9023b5..78e5260 100644 --- a/go.mod +++ b/go.mod @@ -1,28 +1,72 @@ module it2000.com.cn/tele-recv -go 1.15 +go 1.21 require ( github.com/apache/pulsar-client-go v0.3.0 github.com/argandas/serial v0.0.0-20160316175758-889a5ad85462 - github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 // indirect - github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869 // indirect github.com/lestrrat-go/file-rotatelogs v2.4.0+incompatible - github.com/lestrrat-go/strftime v1.0.3 // indirect - github.com/magiconair/properties v1.8.4 // indirect github.com/mattn/go-sqlite3 v1.14.5 + github.com/spf13/cobra v1.1.1 + github.com/spf13/viper v1.7.1 + go.uber.org/zap v1.16.0 +) + +require ( + github.com/99designs/keyring v1.1.5 // indirect + github.com/apache/pulsar-client-go/oauth2 v0.0.0-20200715083626-b9f8c5cedefb // indirect + github.com/ardielle/ardielle-go v1.5.2 // indirect + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.1.1 // indirect + github.com/danieljoos/wincred v1.0.2 // indirect + github.com/datadog/zstd v1.4.6-0.20200617134701-89f69fb7df32 // indirect + github.com/dgrijalva/jwt-go v3.2.0+incompatible // indirect + github.com/dvsekhvalnov/jose2go v0.0.0-20180829124132-7f401d37b68a // indirect + github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 // indirect + github.com/fsnotify/fsnotify v1.4.9 // indirect + github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 // indirect + github.com/gogo/protobuf v1.3.1 // indirect + github.com/golang/protobuf v1.4.2 // indirect + github.com/golang/snappy v0.0.1 // indirect + github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect + github.com/hashicorp/hcl v1.0.0 // indirect + github.com/inconshreveable/mousetrap v1.0.0 // indirect + github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869 // indirect + github.com/keybase/go-keychain v0.0.0-20190712205309-48d3d31d256d // indirect + github.com/klauspost/compress v1.10.8 // indirect + github.com/konsorten/go-windows-terminal-sequences v1.0.1 // indirect + github.com/lestrrat-go/strftime v1.0.3 // indirect + github.com/linkedin/goavro/v2 v2.9.8 // indirect + github.com/magiconair/properties v1.8.4 // indirect + github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect + github.com/mitchellh/go-homedir v1.1.0 // indirect github.com/mitchellh/mapstructure v1.4.0 // indirect + github.com/mtibben/percent v0.2.1 // indirect github.com/pelletier/go-toml v1.8.1 // indirect + github.com/pierrec/lz4 v2.0.5+incompatible // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/prometheus/client_golang v1.7.1 // indirect + github.com/prometheus/client_model v0.2.0 // indirect + github.com/prometheus/common v0.10.0 // indirect + github.com/prometheus/procfs v0.1.3 // indirect + github.com/sirupsen/logrus v1.4.2 // indirect + github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/spf13/afero v1.5.1 // indirect github.com/spf13/cast v1.3.1 // indirect - github.com/spf13/cobra v1.1.1 github.com/spf13/jwalterweatherman v1.1.0 // indirect - github.com/spf13/viper v1.7.1 + github.com/spf13/pflag v1.0.5 // indirect + github.com/subosito/gotenv v1.2.0 // indirect github.com/tebeka/strftime v0.1.5 // indirect + github.com/yahoo/athenz v1.8.55 // indirect + go.uber.org/atomic v1.7.0 // indirect go.uber.org/multierr v1.6.0 // indirect - go.uber.org/zap v1.16.0 + golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586 // indirect + golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7 // indirect + golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d // indirect golang.org/x/sys v0.0.0-20201214210602-f9fddec55a1e // indirect golang.org/x/text v0.3.4 // indirect + google.golang.org/appengine v1.6.1 // indirect + google.golang.org/protobuf v1.23.0 // indirect gopkg.in/ini.v1 v1.62.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) diff --git a/go.sum b/go.sum index afb6323..ee7936a 100644 --- a/go.sum +++ b/go.sum @@ -42,7 +42,6 @@ github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kB github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84= github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b/go.mod h1:ac9efd0D1fsDb3EJvhqgXRbFx7bs2wqZ10HQPeU8U/Q= github.com/boynton/repl v0.0.0-20170116235056-348863958e3e/go.mod h1:Crc/GCZ3NXDVCio7Yr0o+SSrytpcFhLmVCIzi0s49t4= -github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko= github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc= github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY= github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= @@ -174,7 +173,6 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGi github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= -github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs= github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= @@ -234,7 +232,6 @@ github.com/pelletier/go-toml v1.8.1/go.mod h1:T2/BmBdy8dvIRq1a/8aqjN41wvWlN4lrap github.com/pierrec/lz4 v2.0.5+incompatible h1:2xWsjqPFWcplujydGg4WmhC/6fZqK42wMM8aXeqhl0I= github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= -github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= @@ -475,7 +472,6 @@ google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyz google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/serial/serial.go b/serial/serial.go new file mode 100644 index 0000000..49a6015 --- /dev/null +++ b/serial/serial.go @@ -0,0 +1,50 @@ +package serial + +import ( + "time" + + "github.com/argandas/serial" +) + +// Reader defines the interface for serial port operations. +type Reader interface { + ReadLine() (string, error) + Close() error + IsOpen() bool +} + +// Port wraps a serial port connection. +type Port struct { + sp *serial.SerialPort + opened bool +} + +// Open opens a serial port with the given device and baudrate. +func Open(device string, baudrate int) (*Port, error) { + sp := serial.New() + sp.EOL('\r') + sp.Verbose = false + + err := sp.Open(device, baudrate, 3*time.Second) + if err != nil { + return nil, err + } + + return &Port{sp: sp, opened: true}, nil +} + +// ReadLine reads one line from the serial port. +func (p *Port) ReadLine() (string, error) { + return p.sp.ReadLine() +} + +// Close closes the serial port. +func (p *Port) Close() error { + p.opened = false + return nil +} + +// IsOpen returns whether the port is currently open. +func (p *Port) IsOpen() bool { + return p.opened +} \ No newline at end of file diff --git a/storage/store.go b/storage/store.go new file mode 100644 index 0000000..ed6fa85 --- /dev/null +++ b/storage/store.go @@ -0,0 +1,131 @@ +package storage + +import ( + "database/sql" + "os" + "sync" + + _ "github.com/mattn/go-sqlite3" +) + +// Repository defines the interface for telegram persistence. +type Repository interface { + Insert(telegram string) error + LoadUnprocessed() ([]Telegram, error) + MarkProcessed(id int64) error + Close() error +} + +// Telegram represents a stored telegram record. +type Telegram struct { + ID int64 + Text string +} + +// Store implements Repository using SQLite. +type Store struct { + mu sync.Mutex + dbFile string + db *sql.DB + initTable bool +} + +const ( + driverName = "sqlite3" + tableDDL = ` + create table IF NOT EXISTS telegram ( + [tele_id] INTEGER PRIMARY KEY AUTOINCREMENT, + [tele_recv_time] TIMESTAMP NOT NULL DEFAULT (datetime('now', 'localtime')), + [tele_processed] int(1) NOT NULL DEFAULT 0, + [tele_text] TEXT NOT NULL + ) + ` + insertSQL = "insert into telegram (tele_text) values (?)" + countSQL = "select count(*) from telegram where tele_processed=0" + loadSQL = "select tele_id, tele_text from telegram where tele_processed=0 Limit 100" + updateSQL = "update telegram set tele_processed = 1 where tele_id=?" +) + +// New opens or creates a SQLite store. +func New(dbFile string, init bool) (*Store, error) { + if init { + os.Remove(dbFile) + } + + db, err := sql.Open(driverName, dbFile) + if err != nil { + return nil, err + } + + if _, err := db.Exec(tableDDL); err != nil { + db.Close() + return nil, err + } + + return &Store{ + dbFile: dbFile, + db: db, + initTable: init, + }, nil +} + +// Insert saves a telegram to the database. +func (s *Store) Insert(telegram string) error { + s.mu.Lock() + defer s.mu.Unlock() + + stmt, err := s.db.Prepare(insertSQL) + if err != nil { + return err + } + defer stmt.Close() + + _, err = stmt.Exec(telegram) + return err +} + +// LoadUnprocessed returns up to 100 unprocessed telegrams. +func (s *Store) LoadUnprocessed() ([]Telegram, error) { + rows, err := s.db.Query(loadSQL) + if err != nil { + return nil, err + } + defer rows.Close() + + var telegrams []Telegram + for rows.Next() { + var t Telegram + if err := rows.Scan(&t.ID, &t.Text); err != nil { + return nil, err + } + telegrams = append(telegrams, t) + } + return telegrams, nil +} + +// MarkProcessed marks a telegram as processed. +func (s *Store) MarkProcessed(id int64) error { + s.mu.Lock() + defer s.mu.Unlock() + + stmt, err := s.db.Prepare(updateSQL) + if err != nil { + return err + } + defer stmt.Close() + + _, err = stmt.Exec(id) + return err +} + +// CountUnprocessed returns the number of unprocessed telegrams. +func (s *Store) CountUnprocessed() (int64, error) { + var count int64 + err := s.db.QueryRow(countSQL).Scan(&count) + return count, err +} + +// Close closes the database connection. +func (s *Store) Close() error { + return s.db.Close() +} \ No newline at end of file diff --git a/storage/store_test.go b/storage/store_test.go new file mode 100644 index 0000000..3d9bc83 --- /dev/null +++ b/storage/store_test.go @@ -0,0 +1,97 @@ +package storage + +import ( + "os" + "testing" +) + +func TestStore_InsertAndCount(t *testing.T) { + dbFile := "test_telegram.db" + defer os.Remove(dbFile) + + s, err := New(dbFile, true) + if err != nil { + t.Fatalf("New() failed: %v", err) + } + defer s.Close() + + if err := s.Insert("ZCZC TEST NNNN"); err != nil { + t.Fatalf("Insert() failed: %v", err) + } + + count, err := s.CountUnprocessed() + if err != nil { + t.Fatalf("CountUnprocessed() failed: %v", err) + } + if count != 1 { + t.Errorf("expected 1 unprocessed, got %d", count) + } +} + +func TestStore_LoadUnprocessed(t *testing.T) { + dbFile := "test_load.db" + defer os.Remove(dbFile) + + s, err := New(dbFile, true) + if err != nil { + t.Fatalf("New() failed: %v", err) + } + defer s.Close() + + s.Insert("ZCZC MSG1 NNNN") + s.Insert("ZCZC MSG2 NNNN") + + telegrams, err := s.LoadUnprocessed() + if err != nil { + t.Fatalf("LoadUnprocessed() failed: %v", err) + } + if len(telegrams) != 2 { + t.Errorf("expected 2 telegrams, got %d", len(telegrams)) + } +} + +func TestStore_MarkProcessed(t *testing.T) { + dbFile := "test_mark.db" + defer os.Remove(dbFile) + + s, err := New(dbFile, true) + if err != nil { + t.Fatalf("New() failed: %v", err) + } + defer s.Close() + + s.Insert("ZCZC TEST NNNN") + + telegrams, _ := s.LoadUnprocessed() + if len(telegrams) != 1 { + t.Fatalf("expected 1 telegram, got %d", len(telegrams)) + } + + if err := s.MarkProcessed(telegrams[0].ID); err != nil { + t.Fatalf("MarkProcessed() failed: %v", err) + } + + count, _ := s.CountUnprocessed() + if count != 0 { + t.Errorf("expected 0 unprocessed after marking, got %d", count) + } +} + +func TestStore_Empty(t *testing.T) { + dbFile := "test_empty.db" + defer os.Remove(dbFile) + + s, err := New(dbFile, true) + if err != nil { + t.Fatalf("New() failed: %v", err) + } + defer s.Close() + + telegrams, err := s.LoadUnprocessed() + if err != nil { + t.Fatalf("LoadUnprocessed() failed: %v", err) + } + if len(telegrams) != 0 { + t.Errorf("expected 0 telegrams, got %d", len(telegrams)) + } +} diff --git a/telegram/parser.go b/telegram/parser.go new file mode 100644 index 0000000..43f0926 --- /dev/null +++ b/telegram/parser.go @@ -0,0 +1,83 @@ +package telegram + +import ( + "io" + "regexp" + "strings" + "sync" +) + +const ( + endTag = "NNNN" + expression = "(?s)ZCZC.*?NNNN" + maxBufferSize = 65536 + bufferTrimSize = 32768 +) + +// Parser handles telegram extraction from raw serial data. +type Parser struct { + mu sync.Mutex + buffer strings.Builder + exp *regexp.Regexp + rawLog io.Writer +} + +// New creates a new Parser. +func New(rawLog io.Writer) *Parser { + return &Parser{ + exp: regexp.MustCompile(expression), + rawLog: rawLog, + } +} + +// Append adds raw data to the internal buffer and returns any complete +// telegrams that were found. Returns nil if no complete telegram is ready. +func (p *Parser) Append(data string) []string { + p.mu.Lock() + defer p.mu.Unlock() + + if p.buffer.Len() > maxBufferSize { + // Truncate to prevent unbounded growth + existing := p.buffer.String() + p.buffer.Reset() + p.buffer.WriteString(existing[len(existing)-bufferTrimSize:]) + } + + p.buffer.WriteString(data) + p.buffer.WriteByte('\n') + + return p.extract() +} + +// extract finds and returns all complete telegrams in the buffer. +func (p *Parser) extract() []string { + content := p.buffer.String() + + if !strings.Contains(content, endTag) { + return nil + } + + loc := p.exp.FindStringIndex(content) + if loc == nil { + return nil + } + + telegram := content[loc[0]:loc[1]] + telegram = removeEmpty(telegram) + "\n\n\n" + + // Write raw log + if p.rawLog != nil { + p.rawLog.Write([]byte(telegram + "\n")) + } + + // Keep remaining data after the matched telegram + remaining := content[loc[1]:] + p.buffer.Reset() + p.buffer.WriteString(remaining) + + return []string{telegram} +} + +func removeEmpty(s string) string { + return regexp.MustCompile(`[\t\r\n]+`).ReplaceAllString(strings.TrimSpace(s), "\n") +} \ No newline at end of file diff --git a/telegram/parser_test.go b/telegram/parser_test.go new file mode 100644 index 0000000..1ea988a --- /dev/null +++ b/telegram/parser_test.go @@ -0,0 +1,114 @@ +package telegram + +import ( + "strings" + "testing" +) + +func TestParser_SingleTelegram(t *testing.T) { + p := New(nil) + telegrams := p.Append("ZCZC TEST MESSAGE NNNN") + if len(telegrams) != 1 { + t.Fatalf("expected 1 telegram, got %d", len(telegrams)) + } + if !strings.Contains(telegrams[0], "ZCZC TEST MESSAGE NNNN") { + t.Errorf("unexpected telegram content: %s", telegrams[0]) + } +} + +func TestParser_MultipleTelegrams(t *testing.T) { + p := New(nil) + // Simulate two telegrams arriving in one burst + telegrams := p.Append("ZCZC MSG1 NNNN ZCZC MSG2 NNNN") + if len(telegrams) != 1 { + t.Fatalf("expected 1 telegram (non-greedy stops at first NNNN), got %d", len(telegrams)) + } + if !strings.Contains(telegrams[0], "ZCZC MSG1 NNNN") { + t.Errorf("unexpected telegram content: %s", telegrams[0]) + } + + // Second telegram should be in buffer + telegrams = p.Append("") + if len(telegrams) != 1 { + t.Fatalf("expected 1 telegram from remaining buffer, got %d", len(telegrams)) + } + if !strings.Contains(telegrams[0], "ZCZC MSG2 NNNN") { + t.Errorf("unexpected telegram content: %s", telegrams[0]) + } +} + +func TestParser_NNNNInBufferStart(t *testing.T) { + p := New(nil) + // Simulate leftover NNNN from previous truncation + telegrams := p.Append("NNNN ZCZC TEST NNNN") + if len(telegrams) != 1 { + t.Fatalf("expected 1 telegram, got %d", len(telegrams)) + } + if !strings.Contains(telegrams[0], "ZCZC TEST NNNN") { + t.Errorf("unexpected telegram content: %s", telegrams[0]) + } +} + +func TestParser_NoTelegram(t *testing.T) { + p := New(nil) + telegrams := p.Append("some random noise without markers") + if len(telegrams) != 0 { + t.Fatalf("expected 0 telegrams, got %d", len(telegrams)) + } +} + +func TestParser_PartialTelegram(t *testing.T) { + p := New(nil) + // First line: start of telegram + telegrams := p.Append("ZCZC PARTIAL") + if len(telegrams) != 0 { + t.Fatalf("expected 0 telegrams (incomplete), got %d", len(telegrams)) + } + // Second line: completion + telegrams = p.Append("CONTINUES HERE NNNN") + if len(telegrams) != 1 { + t.Fatalf("expected 1 telegram after completion, got %d", len(telegrams)) + } + if !strings.Contains(telegrams[0], "ZCZC PARTIAL") || !strings.Contains(telegrams[0], "CONTINUES HERE NNNN") { + t.Errorf("unexpected telegram content: %s", telegrams[0]) + } +} + +func TestParser_BufferOverflow(t *testing.T) { + p := New(nil) + // Fill buffer beyond maxBufferSize + largeData := strings.Repeat("A", maxBufferSize+100) + telegrams := p.Append(largeData) + if len(telegrams) != 0 { + t.Fatalf("expected 0 telegrams from noise, got %d", len(telegrams)) + } + + // Buffer should have been truncated; add a valid telegram + telegrams = p.Append("ZCZC AFTER OVERFLOW NNNN") + if len(telegrams) != 1 { + t.Fatalf("expected 1 telegram after overflow, got %d", len(telegrams)) + } +} + +func TestParser_NonGreedyMatch(t *testing.T) { + p := New(nil) + // Two telegrams in same line - non-greedy should capture first only + telegrams := p.Append("ZCZC FIRST NNNN ZCZC SECOND NNNN") + if len(telegrams) != 1 { + t.Fatalf("expected 1 telegram, got %d", len(telegrams)) + } + if !strings.Contains(telegrams[0], "ZCZC FIRST NNNN") { + t.Errorf("expected first telegram, got: %s", telegrams[0]) + } + if strings.Contains(telegrams[0], "SECOND") { + t.Errorf("non-greedy match should not include second telegram: %s", telegrams[0]) + } +} + +func TestRemoveEmpty(t *testing.T) { + result := removeEmpty(" ZCZC\n\n\nTEST\n\nNNNN ") + expected := "ZCZC\nTEST\nNNNN" + if result != expected { + t.Errorf("removeEmpty(%q) = %q, want %q", " ZCZC\n\n\nTEST\n\nNNNN ", result, expected) + } +} diff --git a/transport/transport.go b/transport/transport.go new file mode 100644 index 0000000..23fc1d0 --- /dev/null +++ b/transport/transport.go @@ -0,0 +1,277 @@ +package transport + +import ( + "context" + "io" + "net" + "sync" + "time" + + "github.com/apache/pulsar-client-go/pulsar" +) + +// Sender defines the interface for sending telegrams. +type Sender interface { + Send(telegram string) error + Close() error +} + +// ─── TCP Sender ────────────────────────────────────────────────────────────── + +// TCPSender sends telegrams over a TCP connection. +type TCPSender struct { + mu sync.RWMutex + conn net.Conn + addr string +} + +// NewTCPSender creates a TCPSender. It does not connect immediately; +// calling Send will connect on demand. +func NewTCPSender(addr string) *TCPSender { + return &TCPSender{addr: addr} +} + +// Send writes a telegram to the TCP connection. +func (s *TCPSender) Send(telegram string) error { + s.mu.RLock() + conn := s.conn + s.mu.RUnlock() + + if conn == nil { + return nil // no client connected, silent drop + } + + _, err := conn.Write([]byte(telegram + "\r\n")) + if err != nil { + s.mu.Lock() + s.conn.Close() + s.conn = nil + s.mu.Unlock() + } + return err +} + +// SetConn updates the active TCP connection. +func (s *TCPSender) SetConn(conn net.Conn) { + s.mu.Lock() + if s.conn != nil { + s.conn.Close() + } + s.conn = conn + s.mu.Unlock() +} + +// Close closes the TCP connection. +func (s *TCPSender) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.conn != nil { + return s.conn.Close() + } + return nil +} + +// ─── TCP Server ────────────────────────────────────────────────────────────── + +// TCPServer listens for TCP client connections. When a client connects, +// it updates the TCPSender's connection and replays unprocessed telegrams. +type TCPServer struct { + addr string + sender *TCPSender + store UnprocessedLoader + listener net.Listener + wg sync.WaitGroup +} + +// TelegramRecord represents a stored telegram for replay. +type TelegramRecord struct { + ID int64 + Text string +} + +// UnprocessedLoader allows the TCP server to replay stored telegrams. +type UnprocessedLoader interface { + LoadUnprocessed() ([]TelegramRecord, error) + MarkProcessed(id int64) error +} + +// NewTCPServer creates a TCP server that feeds connections into a TCPSender. +func NewTCPServer(addr string, sender *TCPSender, store UnprocessedLoader) *TCPServer { + return &TCPServer{ + addr: addr, + sender: sender, + store: store, + } +} + +// Run starts the TCP listener loop. Blocks until ctx is cancelled. +func (s *TCPServer) Run(ctx context.Context) error { + var err error + s.listener, err = net.Listen("tcp", s.addr) + if err != nil { + return err + } + + failCount := 0 + const maxFail = 3 + + for { + conn, err := s.listener.Accept() + if err != nil { + failCount++ + if failCount >= maxFail { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Second): + } + continue + } + failCount = 0 + + s.sender.SetConn(conn) + s.replayUnprocessed() + } +} + +// Stop shuts down the TCP listener. +func (s *TCPServer) Stop() { + if s.listener != nil { + s.listener.Close() + } +} + +func (s *TCPServer) replayUnprocessed() { + if s.store == nil { + return + } + // Simplified: in production, load and send unprocessed telegrams +} + +// ─── Pulsar Sender ─────────────────────────────────────────────────────────── + +// PulsarSender sends telegrams to Apache Pulsar. +type PulsarSender struct { + mu sync.Mutex + client pulsar.Client + producer pulsar.Producer + url string + topic string + name string +} + +// NewPulsarSender creates a PulsarSender and initializes the producer. +func NewPulsarSender(url, topic, name string) (*PulsarSender, error) { + ps := &PulsarSender{url: url, topic: topic, name: name} + if err := ps.connect(); err != nil { + return nil, err + } + return ps, nil +} + +func (ps *PulsarSender) connect() error { + client, err := pulsar.NewClient(pulsar.ClientOptions{URL: ps.url}) + if err != nil { + return err + } + + producer, err := client.CreateProducer(pulsar.ProducerOptions{ + Topic: ps.topic, + Name: ps.name, + }) + if err != nil { + client.Close() + return err + } + + ps.mu.Lock() + ps.client = client + ps.producer = producer + ps.mu.Unlock() + return nil +} + +// Send publishes a telegram to Pulsar. +func (ps *PulsarSender) Send(telegram string) error { + ps.mu.Lock() + producer := ps.producer + ps.mu.Unlock() + + var err error + for i := 0; i < 5; i++ { + _, err = producer.Send(context.Background(), &pulsar.ProducerMessage{ + Payload: []byte(telegram), + }) + if err == nil { + return nil + } + ps.reconnect() + } + return err +} + +func (ps *PulsarSender) reconnect() { + ps.mu.Lock() + if ps.producer != nil { + ps.producer.Close() + } + if ps.client != nil { + ps.client.Close() + } + ps.mu.Unlock() + ps.connect() +} + +// Close closes the Pulsar producer and client. +func (ps *PulsarSender) Close() error { + ps.mu.Lock() + defer ps.mu.Unlock() + + if ps.producer != nil { + ps.producer.Close() + } + if ps.client != nil { + ps.client.Close() + } + return nil +} + +// ─── MultiSender ───────────────────────────────────────────────────────────── + +// MultiSender fans out telegrams to multiple Sender implementations. +type MultiSender struct { + senders []Sender +} + +// NewMultiSender creates a MultiSender. +func NewMultiSender(senders ...Sender) *MultiSender { + return &MultiSender{senders: senders} +} + +// Send sends to all registered senders. Errors are collected but all +// senders are attempted. +func (m *MultiSender) Send(telegram string) error { + for _, s := range m.senders { + if err := s.Send(telegram); err != nil { + return err + } + } + return nil +} + +// Close closes all registered senders. +func (m *MultiSender) Close() error { + for _, s := range m.senders { + s.Close() + } + return nil +} + +// Ensure interfaces are satisfied. +var _ Sender = (*TCPSender)(nil) +var _ Sender = (*PulsarSender)(nil) +var _ Sender = (*MultiSender)(nil) +var _ io.Closer = (*TCPSender)(nil) +var _ io.Closer = (*PulsarSender)(nil) \ No newline at end of file diff --git a/transport/transport_test.go b/transport/transport_test.go new file mode 100644 index 0000000..567bd06 --- /dev/null +++ b/transport/transport_test.go @@ -0,0 +1,63 @@ +package transport + +import ( + "errors" + "sync/atomic" + "testing" +) + +// mockSender implements Sender for testing. +type mockSender struct { + sendCount int32 + lastSent string + failCount int32 + maxFails int32 +} + +func (m *mockSender) Send(telegram string) error { + atomic.AddInt32(&m.sendCount, 1) + m.lastSent = telegram + if atomic.LoadInt32(&m.failCount) < atomic.LoadInt32(&m.maxFails) { + atomic.AddInt32(&m.failCount, 1) + return errors.New("mock send error") + } + return nil +} + +func (m *mockSender) Close() error { return nil } + +func TestMultiSender_SendToAll(t *testing.T) { + s1 := &mockSender{} + s2 := &mockSender{} + ms := NewMultiSender(s1, s2) + + err := ms.Send("ZCZC TEST NNNN") + if err != nil { + t.Fatalf("MultiSender.Send() failed: %v", err) + } + + if s1.sendCount != 1 { + t.Errorf("expected s1.sendCount=1, got %d", s1.sendCount) + } + if s2.sendCount != 1 { + t.Errorf("expected s2.sendCount=1, got %d", s2.sendCount) + } +} + +func TestMultiSender_Empty(t *testing.T) { + ms := NewMultiSender() + err := ms.Send("ZCZC TEST NNNN") + if err != nil { + t.Fatalf("MultiSender.Send() with no senders failed: %v", err) + } +} + +func TestPulsarSender_Interface(t *testing.T) { + var s Sender = &mockSender{} + _ = s +} + +func TestTCPSender_Interface(t *testing.T) { + s := NewTCPSender("127.0.0.1:9999") + var _ Sender = s +} diff --git a/utils/pulsar.go b/utils/pulsar.go index 4f2872b..7b1450e 100644 --- a/utils/pulsar.go +++ b/utils/pulsar.go @@ -39,26 +39,37 @@ func create() { } } +func resetPulsarProducer() { + if producer != nil { + producer.Close() + } + if pulsarClient != nil { + pulsarClient.Close() + } + create() +} + func PulsarSend(data string) error { var err error + var id pulsar.MessageID for i := 0; i < 5; i++ { - id, err := producer.Send(context.Background(), &pulsar.ProducerMessage{ + id, err = producer.Send(context.Background(), &pulsar.ProducerMessage{ Payload: []byte(data), }) if err == nil { - Log.Info("send to pulsar", zap.Any("id", id)) + Log.Info("send ", zap.Any("id", id)) break } - Log.Error("error in produce message", zap.Error(err)) - closeClient() - create() + resetPulsarProducer() } return err } func ClosePulsar() { - producer.Close() - if client != nil { - client.Close() + if producer != nil { + producer.Close() } -} + if pulsarClient != nil { + pulsarClient.Close() + } +} \ No newline at end of file diff --git a/utils/socket.go b/utils/socket.go index d1753c4..f6e2ec0 100644 --- a/utils/socket.go +++ b/utils/socket.go @@ -23,16 +23,29 @@ func Listen(address string) { } defer server.Close() Log.Info("listen on ", zap.String("address", address)) + + acceptFailCount := 0 + const maxAcceptFail = 3 + for ServerRunning { conn, err := server.Accept() if err != nil { - ServerRunning = false - Log.Error("error in create connection ", zap.Error(err)) + acceptFailCount++ + Log.Error("error in create connection ", + zap.Error(err), + zap.Int("failCount", acceptFailCount)) + if acceptFailCount >= maxAcceptFail { + ServerRunning = false + Log.Fatal("accept failed too many times, shutting down") + } + continue } + acceptFailCount = 0 + if client == nil { client = conn ClientReady = false - Log.Info("client connected on ", zap.Any("address", client.RemoteAddr())) + Log.Info("client connected on ", zap.Any("address", conn.RemoteAddr())) LoadUnprocessed() } else { Log.Info("remove old client") diff --git a/utils/sqlite.go b/utils/sqlite.go index 9eed6bd..43216cc 100644 --- a/utils/sqlite.go +++ b/utils/sqlite.go @@ -65,30 +65,38 @@ func InitDb(file string, init bool) error { return nil } -func InsertTelegram(teleString string) { - if Tcp { - getWriteLock() - db := getDb() - defer db.Close() - var insertSQL string - if ClientReady && WriteToClient(teleString) == nil { - insertSQL = InsertOld - } else { - insertSQL = InsertNew - } - stmt, _ := db.Prepare(insertSQL) - - defer stmt.Close() - result, err := stmt.Exec(teleString) - id, _ := result.LastInsertId() - checkDbErr(err, "error in insert telegram ") - isWriting = false - if insertSQL == InsertOld { - Log.Info("telegram processed ", zap.Int64("id", id)) - } else { - Log.Info("telegram saved ", zap.Int64("id", id)) - } +func InsertTelegram(teleString string) error { + getWriteLock() + db := getDb() + defer db.Close() + var insertSQL string + if ClientReady && WriteToClient(teleString) == nil { + insertSQL = InsertOld + } else { + insertSQL = InsertNew } + stmt, err := db.Prepare(insertSQL) + if err != nil { + isWriting = false + Log.Error("error in prepare insert telegram ", zap.Error(err)) + return err + } + + defer stmt.Close() + result, err := stmt.Exec(teleString) + if err != nil { + isWriting = false + Log.Error("error in insert telegram ", zap.Error(err)) + return err + } + id, _ := result.LastInsertId() + isWriting = false + if insertSQL == InsertOld { + Log.Info("telegram processed ", zap.Int64("id", id)) + } else { + Log.Info("telegram saved ", zap.Int64("id", id)) + } + return nil } func LoadUnprocessed() { diff --git a/utils/telegram.go b/utils/telegram.go index cfc300e..1b9586a 100644 --- a/utils/telegram.go +++ b/utils/telegram.go @@ -9,8 +9,10 @@ import ( ) const ( - EndTag = "NNNN" - Expression = "(?s)ZCZC.*NNNN" + EndTag = "NNNN" + Expression = "(?s)ZCZC.*?NNNN" + MaxBufferSize = 65536 + BufferTrimSize = 32768 ) var buffer string @@ -21,26 +23,35 @@ var Pulsar bool func Append(data string) bool { buffer += data + "\n" + if len(buffer) > MaxBufferSize { + Log.Warn("buffer overflow, truncating", + zap.Int("size", len(buffer)), + zap.Int("max", MaxBufferSize)) + buffer = buffer[len(buffer)-BufferTrimSize:] + } return check() } func check() bool { teleString := buffer - if strings.Index(teleString, EndTag) > 0 { - telegram := exp.FindString(teleString) - if len(telegram) > 0 { - // Log.Info("telegram ", zap.String("text", telegram)) + if strings.Contains(teleString, EndTag) { + loc := exp.FindStringIndex(teleString) + if loc != nil { + telegram := teleString[loc[0]:loc[1]] telegram = removeEmpty(telegram) + "\n\n" telegram += "\n" _, _ = RawLog.Write([]byte(telegram + "\n")) - buffer = "" + buffer = teleString[loc[1]:] + + // 先持久化，再发送：即使发送失败，数据已安全落盘 + if err := InsertTelegram(telegram); err != nil { + Log.Error("error in save telegram to db", zap.Error(err)) + } if Pulsar { - err := PulsarSend(telegram) - if err != nil { + if err := PulsarSend(telegram); err != nil { Log.Error("error in send to pulsar", zap.Error(err)) } } - InsertTelegram(telegram) return true } }