windy/tele-recv
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: comprehensive bug fixes and architecture restructuring

Browse Source 
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
This commit is contained in:
w1ndyb0y
2026-07-10 15:32:34 +08:00
parent 5f3334ec87
commit 419f25f0dd
20 changed files with 1519 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Taskfile.yml
+150
View File
@@ -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
app/app.go
+268
View File
@@ -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
}
cmd/root.go
+35 -59
View File
@@ -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)
}
cmd/start.go
+14 -80
View File
@@ -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()
}
cmd/test.go
-10
View File
@@ -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")
}
config/config.go
+43
View File
@@ -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
}
config/loader.go
+54
View File
@@ -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
}
go.mod
+52 -8
View File
@@ -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
)
go.sum
-4
View File
@@ -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=
serial/serial.go
+50
View File
@@ -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
}
storage/store.go
+131
View File
@@ -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()
}
storage/store_test.go
+97
View File
@@ -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))
}
}
telegram/parser.go
+83
View File
@@ -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")
}
telegram/parser_test.go
+114
View File
@@ -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)
}
}
transport/transport.go
+277
View File
@@ -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)
transport/transport_test.go
+63
View File
@@ -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
}
utils/pulsar.go
+20 -9
View File
@@ -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()
}
}
utils/socket.go
+16 -3
View File
@@ -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")
utils/sqlite.go
+31 -23
View File
@@ -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() {
utils/telegram.go
+21 -10
View File
@@ -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
}
}