diff --git a/.drone.yml b/.drone.yml new file mode 100644 index 0000000..ec15392 --- /dev/null +++ b/.drone.yml @@ -0,0 +1,52 @@ +--- +kind: pipeline +name: default +type: docker + +platform: + os: linux + arch: amd64 + +clone: + image: reg.int.it2000.com.cn/devops/git + tags: true + skip_verify: true + +steps: + - name: build + image: reg.int.it2000.com.cn/library/golang:1.15.6-buster + volumes: + - name: deps + path: /go + environment: + GO111MODULE: on + GOPROXY: https://goproxy.io + commands: + - go build -o tele-recv-linux + + - name: release + image: plugins/gitea-release + volumes: + - name: certs + path: /etc/ssl/certs + settings: + api_key: fe1e1a8dd77de7452a3f473984c7ac6fe6c5abaf + base_url: https://gitea.int.it2000.com.cn + title: 0.0.1 + files: + - tele-recv-linux + checksum: + - sha256 + when: + event: tag + +volumes: + - name: deps + host: + path: /opt/go + - name: certs + host: + path: /etc/ssl/certs + - name: docker + host: + path: /var/run/docker.sock \ No newline at end of file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3c379ef --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +tty* +telegram.db +.idea \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml deleted file mode 100644 index 25d714b..0000000 --- a/.idea/modules.xml +++ /dev/null @@ -1,8 +0,0 @@ - - - - - - - - \ No newline at end of file diff --git a/.idea/tele-recv.iml b/.idea/tele-recv.iml deleted file mode 100644 index c956989..0000000 --- a/.idea/tele-recv.iml +++ /dev/null @@ -1,8 +0,0 @@ - - - - - - - - \ No newline at end of file diff --git a/.idea/workspace.xml b/.idea/workspace.xml deleted file mode 100644 index 358167b..0000000 --- a/.idea/workspace.xml +++ /dev/null @@ -1,44 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - true - - - - - - - \ No newline at end of file diff --git a/README.md b/README.md index e69de29..9dfb9f3 100644 --- a/README.md +++ b/README.md @@ -0,0 +1 @@ +### 电报接收 \ No newline at end of file diff --git a/cmd/read.go b/cmd/read.go deleted file mode 100644 index d0d2799..0000000 --- a/cmd/read.go +++ /dev/null @@ -1,55 +0,0 @@ -/* -Copyright © 2020 NAME HERE - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ -package cmd - -import ( - "fmt" - - "github.com/spf13/cobra" -) - -// readCmd represents the read command -var readCmd = &cobra.Command{ - Use: "read", - Short: "A brief description of your command", - Long: `A longer description that spans multiple lines and likely contains examples -and usage of using your command. For example: - -Cobra is a CLI library for Go that empowers applications. -This application is a tool to generate the needed files -to quickly create a Cobra application.`, - Run: func(cmd *cobra.Command, args []string) { - fmt.Println("read called") - }, -} - -func init() { - rootCmd.AddCommand(readCmd) - - // Here you will define your flags and configuration settings. - - // Cobra supports Persistent Flags which will work for this command - // and all subcommands, e.g.: - // readCmd.PersistentFlags().String("foo", "", "A help for foo") - - // Cobra supports local flags which will only run when this command - // is called directly, e.g.: - // readCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle") -} - -func read() { - -} diff --git a/cmd/root.go b/cmd/root.go index bf391e3..2db00aa 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -24,10 +24,16 @@ import ( ) var ( - cfgFile string - device string + cfgFile string + device string baudrate int bytesize byte + lograw bool + + dbFile string + dbInit bool + + socketAddress string // rootCmd represents the base command when called without any subcommands rootCmd = &cobra.Command{ @@ -83,12 +89,15 @@ func initConfig() { 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()) device = viper.GetString("serial.device") baudrate = viper.GetInt("serial.baudrate") bytesize = byte(viper.GetInt("serial.bytesize")) + lograw = viper.GetBool("serial.lograw") + dbFile = viper.GetString("sqlite.file") + dbInit = viper.GetBool("sqlite.init") + socketAddress = viper.GetString("socket.address") } } diff --git a/cmd/start.go b/cmd/start.go new file mode 100644 index 0000000..0857a5a --- /dev/null +++ b/cmd/start.go @@ -0,0 +1,106 @@ +/* +Copyright © 2020 NAME HERE + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ +package cmd + +import ( + "fmt" + "log" + "os" + "os/signal" + "syscall" + + "it2000.com.cn/tele-recv/utils" + + "github.com/spf13/cobra" +) + +// startCmd represents the start command +var startCmd = &cobra.Command{ + Use: "start", + Short: "start telegram receive service", + Long: `open serial port + start a tcp server for processing`, + Run: func(cmd *cobra.Command, args []string) { + 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 + fmt.Println() + fmt.Println(sig) + utils.ServerRunning = false + utils.StopSocketServer() + 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. + fmt.Println("awaiting signal") + _ = utils.InitDb(dbFile, dbInit) + go utils.Listen(socketAddress) + for utils.ServerRunning { + if !utils.IsPortOpen() { + log.Println("try to open port") + err := utils.OpenPort(device, baudrate, bytesize) + if err != nil { + log.Fatal("error in open serial port ", err) + } + log.Println("starting read") + } + buffer, count := utils.ReadPort() + if count > 0 { + if lograw { + log.Print(string(buffer)) + } + utils.Append(buffer) + } + } + <-done + fmt.Println("exiting") + +} diff --git a/cmd/test.go b/cmd/test.go index c7da629..18c2391 100644 --- a/cmd/test.go +++ b/cmd/test.go @@ -16,32 +16,33 @@ limitations under the License. package cmd import ( - "fmt" "github.com/spf13/cobra" - "github.com/tarm/serial" - "os" + "it2000.com.cn/tele-recv/utils" + "log" ) // testCmd represents the test command var testCmd = &cobra.Command{ Use: "test", - Short: "Test Config", - Long: `Test config file. -Test serial port`, + Short: "Test running environment", + Long: `loading config file +Test serial port +Test load sqlite database and init table`, Run: func(cmd *cobra.Command, args []string) { - fmt.Println("test called") - fmt.Println("device :",device) - c := &serial.Config{Name: device, Baud: baudrate, Size: bytesize} - s, err := serial.OpenPort(c) - if err != nil { - fmt.Println(err) - os.Exit(1) - } - fmt.Println(c , "opened") - _ = s.Close() + test() }, } +func test() { + log.Println("Testing environment") + _ = utils.OpenPort(device, baudrate, bytesize) + + err := utils.InitDb(dbFile, dbInit) + if err != nil { + println("create database error ", err) + } +} + func init() { rootCmd.AddCommand(testCmd) diff --git a/database/sqlite.go b/database/sqlite.go deleted file mode 100644 index 6fb0296..0000000 --- a/database/sqlite.go +++ /dev/null @@ -1,41 +0,0 @@ -package database - -import ( - "database/sql" - "log" - "os" -) - -const ( - DbDriver = "sqlite3" - TableCreate =`''' - 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 - ) - ''' -` -) -var ( - db *sql.DB -) - -func getDb(file string, init bool) (*sql.DB, error) { - log.Println("init database ", init) - if init { - log.Println("remove database file", file) - os.Remove(file) - } - db, err := sql.Open(DbDriver, file) - defer db.Close() - if err != nil { - log.Fatal("error in open database ", err) - } - _, err = db.Exec(TableCreate) - if err != nil { - log.Fatal("error in create telegram table", err) - } - return db, nil -} \ No newline at end of file diff --git a/go.mod b/go.mod index 80c8d48..4f116c0 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.15 require ( github.com/fsnotify/fsnotify v1.4.9 // indirect github.com/magiconair/properties v1.8.4 // indirect + github.com/mattn/go-sqlite3 v1.14.5 github.com/mitchellh/mapstructure v1.4.0 // indirect github.com/pelletier/go-toml v1.8.1 // indirect github.com/spf13/afero v1.5.1 // indirect diff --git a/go.sum b/go.sum index 0620a98..41b0411 100644 --- a/go.sum +++ b/go.sum @@ -118,6 +118,8 @@ github.com/magiconair/properties v1.8.4 h1:8KGKTcQQGm0Kv7vEbKFErAoAOFyyacLStRtQS github.com/magiconair/properties v1.8.4/go.mod h1:y3VJvCyxH9uVvJTWEGAELF3aiYNyPKd5NZ3oSwXrF60= github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU= github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4= +github.com/mattn/go-sqlite3 v1.14.5 h1:1IdxlwTNazvbKJQSxoJ5/9ECbEeaTTyeU7sEAZ5KKTQ= +github.com/mattn/go-sqlite3 v1.14.5/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc= diff --git a/telegram.yaml b/telegram.yaml index 86babc5..56d430a 100644 --- a/telegram.yaml +++ b/telegram.yaml @@ -2,7 +2,13 @@ serial: device: ttyS1 baudrate: 9600 bytesize: 7 + lograw: true sqlite: file: telegram.db + init: true + + +socket: + address: 127.0.0.1:6000 \ No newline at end of file diff --git a/utils/serial.go b/utils/serial.go new file mode 100644 index 0000000..06c1855 --- /dev/null +++ b/utils/serial.go @@ -0,0 +1,48 @@ +package utils + +import ( + "io" + "log" + "time" + + "github.com/tarm/serial" +) + +var ( + device string + baudrate int + bytesize byte + port *serial.Port + portOpened = false +) + +func OpenPort(device string, baudrate int, bytesize byte) error { + log.Println("opening port device:", device, ", baudrate=", baudrate, ", bytesize=", bytesize) + config := &serial.Config{Name: device, Baud: baudrate, Size: bytesize, ReadTimeout: time.Second * 1} + var err error + port, err = serial.OpenPort(config) + if err != nil { + log.Fatalln("error in open serial port ", err) + return err + } + log.Println("opened port ", port) + portOpened = true + return nil +} + +func ReadPort() ([]byte, int) { + buffer := make([]byte, 256) + if !ServerRunning { + return buffer, 0 + } + count, err := port.Read(buffer) + if err != nil && err != io.EOF { + log.Println("error in read port ", err) + //TODO: 3 time fail try to + } + return buffer, count +} + +func IsPortOpen() bool { + return portOpened +} diff --git a/utils/socket.go b/utils/socket.go new file mode 100644 index 0000000..f9d6a9a --- /dev/null +++ b/utils/socket.go @@ -0,0 +1,87 @@ +package utils + +import ( + "io" + "log" + "net" + "time" +) + +const ServerType = "tcp" + +var ServerRunning = false +var client net.Conn +var server net.Listener +var ClientReady = false + +func Listen(address string) { + var err error + server, err = net.Listen(ServerType, address) + if err != nil { + log.Fatal("error in create server ", err) + } + defer server.Close() + log.Println("listen on ", address) + for ServerRunning { + conn, err := server.Accept() + if err != nil { + ServerRunning = false + log.Fatalln("error in create connection ", err) + } + if client == nil { + client = conn + ClientReady = false + log.Println("client connected ", client.RemoteAddr()) + LoadUnprocessed() + } else { + log.Println("client already connected, close") + client.Close() + client = nil + conn.Close() + } + if IsClientConnected() { + connCheck() + } + time.Sleep(2 * time.Second) + } +} + +func WriteToClient(data string) error { + _, err := client.Write([]byte(data + "\r\n")) + if err != nil { + log.Println("error in write data to client socket ", err) + closeClient() + } + return err +} + +func connCheck() bool { + _, err := client.Read(make([]byte, 0)) + if err != nil && err != io.EOF { + // this connection is invalid + log.Println("conn closed....", err) + client = nil + return false + } + return true +} + +func IsClientConnected() bool { + return client != nil +} + +func closeClient() { + if client != nil { + client.Close() + client = nil + } +} + +func StopSocketServer() { + if IsClientConnected() { + client.Close() + } + if server != nil { + server.Close() + } +} diff --git a/utils/sqlite.go b/utils/sqlite.go new file mode 100644 index 0000000..743138e --- /dev/null +++ b/utils/sqlite.go @@ -0,0 +1,176 @@ +package utils + +import ( + "database/sql" + "log" + "os" + "time" + + _ "github.com/mattn/go-sqlite3" +) + +const ( + DbDriver = "sqlite3" + TableCreate = ` + 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 + ) +` + InsertNew = "insert into telegram (tele_text) values (?)" + InsertOld = "insert into telegram (tele_text, tele_processed) values (?, 1)" + + count = "select count(*) from telegram where tele_processed=0" + load = "select tele_id, tele_text from telegram where tele_processed=0 Limit 100" + update = "update telegram set tele_processed = 1 where tele_id=?" +) + +type telegram struct { + id int64 + text string +} + +var ( + dbFile string + initTable bool + initialized bool + isWriting bool +) + +func getDb() *sql.DB { + //log.Println("init database ", initTable) + if initTable && !initialized { + log.Println("remove database file", dbFile) + os.Remove(dbFile) + } + db, err := sql.Open(DbDriver, dbFile) + if err != nil { + log.Fatal("error in open database ", err) + } + return db +} + +func InitDb(file string, init bool) error { + dbFile = file + initTable = init + var dbErr error + db := getDb() + defer db.Close() + _, dbErr = db.Exec(TableCreate) + checkDbErr(dbErr, "error in create table") + log.Println("database ", db, " initialized") + initialized = true + return nil +} + +func InsertTelegram(teleString string) { + 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.Println("telegram processed ", id) + } else { + log.Println("telegram saved ", id) + } +} + +func LoadUnprocessed() { + log.Println("loading telegram") + db := getDb() + defer db.Close() + for { + telegrams := getTelegram(db) + if len(telegrams) < 1 { + break + } + for _, t := range telegrams { + if WriteToClient(t.text) == nil { + processed(db, t) + } + } + } + ClientReady = true + log.Println("client is ready for new telegram") +} + +func getTelegram(db *sql.DB) []telegram { + var telegrams []telegram + num := countTelegram() + log.Println("unprocessed : ", num) + if num < 1 { + return telegrams + } + log.Println("try to load unprocessed telegram ", num) + rows, err := db.Query(load) + checkDbErr(err, "error in query telegram") + defer rows.Close() + for rows.Next() { + var t telegram + err := rows.Scan(&t.id, &t.text) + checkDbErr(err, "error in scan telegram") + telegrams = append(telegrams, t) + } + return telegrams +} + +func processed(db *sql.DB, t telegram) { + getWriteLock() + tx, err := db.Begin() + checkDbErr(err, "error in begin update") + stmt, err := db.Prepare(update) + defer stmt.Close() + checkDbErr(err, "error in create update statement") + result, err := stmt.Exec(t.id) + checkDbErr(err, "error in update telegram status") + id, _ := result.RowsAffected() + log.Println(t.id, " telegram update : ", id > 0) + isWriting = false + log.Println("release lock") + tx.Commit() +} + +func countTelegram() int64 { + db := getDb() + var countErr error + stmt, _ := db.Prepare(count) + defer stmt.Close() + var num int64 + countErr = stmt.QueryRow().Scan(&num) + checkDbErr(countErr, "error in count telegram") + return num +} + +func getWriteLock() { + count := 0 + log.Println("write lock :", isWriting) + for isWriting && count < 10 { + log.Println("waiting for write ") + time.Sleep(5 * time.Second) + count++ + } + if count >= 10 { + log.Fatalln("database locked") + } + isWriting = true +} + +func checkDbErr(err error, msg string) { + if err != nil { + log.Fatalln(msg, err) + } +} diff --git a/utils/telegram.go b/utils/telegram.go new file mode 100644 index 0000000..76935ac --- /dev/null +++ b/utils/telegram.go @@ -0,0 +1,37 @@ +package utils + +import ( + "log" + "regexp" + "strings" +) + +const ( + EndTag = "NNNN" + Expression = "(?s)ZCZC.*NNNN" +) + +var buffer []byte + +func Append(data []byte) { + buffer = append(buffer, data...) + if buffer[len(buffer)-1] == '\r' { + buffer = append(buffer, '\n') + } + check() +} + +func check() { + teleString := string(buffer) + if strings.Index(teleString, EndTag) > 0 { + exp, _ := regexp.Compile(Expression) + telegram := exp.FindString(string(buffer)) + + if len(telegram) > 0 { + log.Println("telegram : ", telegram) + buffer = nil + InsertTelegram(telegram) + } + } + +}