Skip to content

Commit 2823601

Browse files
team-ideteam-ide
committed
添加第三方中间件接口
1 parent cdabeec commit 2823601

14 files changed

+141
-47
lines changed

‎db/iface.go

+4
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
package db
2+
3+
type IService interface {
4+
}

‎db/service.go

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
package db

‎elasticsearch/iface.go

+4
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
package elasticsearch
2+
3+
type IService interface {
4+
}

‎elasticsearch/service.go

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
package elasticsearch

‎kafka/iface.go

+4
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
package kafka
2+
3+
type IService interface {
4+
}

‎kafka/service.go

+26-24
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@ import (
99
"fmt"
1010
"github.com/Shopify/sarama"
1111
"github.com/team-ide/go-tool/util"
12-
"io/ioutil"
1312
"sort"
1413
"strconv"
1514
"strings"
@@ -25,28 +24,31 @@ type Config struct {
2524
}
2625

2726
// New 创建kafka服务
28-
func New(config Config) (*KafkaService, error) {
29-
service := &KafkaService{
27+
func New(config Config) (IService, error) {
28+
service := &Service{
3029
Config: config,
3130
}
3231
err := service.init()
33-
return service, err
32+
if err != nil {
33+
return nil, err
34+
}
35+
return service, nil
3436
}
3537

36-
// KafkaService 注册处理器在线信息等
37-
type KafkaService struct {
38+
// Service 注册处理器在线信息等
39+
type Service struct {
3840
Config
3941
}
4042

41-
func (this_ *KafkaService) init() (err error) {
43+
func (this_ *Service) init() (err error) {
4244
return
4345
}
4446

45-
func (this_ *KafkaService) Stop() {
47+
func (this_ *Service) Stop() {
4648

4749
}
4850

49-
func (this_ *KafkaService) GetServers() []string {
51+
func (this_ *Service) GetServers() []string {
5052
var servers []string
5153
if this_.Address == "" {
5254
return servers
@@ -61,7 +63,7 @@ func (this_ *KafkaService) GetServers() []string {
6163
return servers
6264
}
6365

64-
func (this_ *KafkaService) getClient() (saramaClient sarama.Client, err error) {
66+
func (this_ *Service) getClient() (saramaClient sarama.Client, err error) {
6567
config := sarama.NewConfig()
6668
config.Consumer.Return.Errors = true
6769
config.Consumer.Offsets.Initial = sarama.OffsetOldest
@@ -77,7 +79,7 @@ func (this_ *KafkaService) getClient() (saramaClient sarama.Client, err error) {
7779
if this_.CertPath != "" {
7880
certPool := x509.NewCertPool()
7981
var pemCerts []byte
80-
pemCerts, err = ioutil.ReadFile(this_.CertPath)
82+
pemCerts, err = util.ReadFile(this_.CertPath)
8183
if err != nil {
8284
return
8385
}
@@ -102,7 +104,7 @@ func (this_ *KafkaService) getClient() (saramaClient sarama.Client, err error) {
102104
return
103105
}
104106

105-
func (this_ *KafkaService) Info() (res interface{}, err error) {
107+
func (this_ *Service) Info() (res interface{}, err error) {
106108

107109
return
108110
}
@@ -118,7 +120,7 @@ type TopicInfo struct {
118120
Topic string `json:"topic"`
119121
}
120122

121-
func (this_ *KafkaService) GetTopics() (res []*TopicInfo, err error) {
123+
func (this_ *Service) GetTopics() (res []*TopicInfo, err error) {
122124
var saramaClient sarama.Client
123125
saramaClient, err = this_.getClient()
124126
if err != nil {
@@ -142,7 +144,7 @@ func (this_ *KafkaService) GetTopics() (res []*TopicInfo, err error) {
142144
return
143145
}
144146

145-
func (this_ *KafkaService) Pull(groupId string, topics []string, PullSize int, PullTimeout int, keyType, valueType string) (msgList []*Message, err error) {
147+
func (this_ *Service) Pull(groupId string, topics []string, PullSize int, PullTimeout int, keyType, valueType string) (msgList []*Message, err error) {
146148
if PullSize <= 0 {
147149
PullSize = 10
148150
}
@@ -217,7 +219,7 @@ func (handler *consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi
217219
return nil
218220
}
219221

220-
func (this_ *KafkaService) MarkOffset(groupId string, topic string, partition int32, offset int64) (err error) {
222+
func (this_ *Service) MarkOffset(groupId string, topic string, partition int32, offset int64) (err error) {
221223
var saramaClient sarama.Client
222224
saramaClient, err = this_.getClient()
223225
if err != nil {
@@ -237,7 +239,7 @@ func (this_ *KafkaService) MarkOffset(groupId string, topic string, partition in
237239
return
238240
}
239241

240-
func (this_ *KafkaService) ResetOffset(groupId string, topic string, partition int32, offset int64) (err error) {
242+
func (this_ *Service) ResetOffset(groupId string, topic string, partition int32, offset int64) (err error) {
241243
var saramaClient sarama.Client
242244
saramaClient, err = this_.getClient()
243245
if err != nil {
@@ -259,7 +261,7 @@ func (this_ *KafkaService) ResetOffset(groupId string, topic string, partition i
259261
return
260262
}
261263

262-
func (this_ *KafkaService) CreatePartitions(topic string, count int32) (err error) {
264+
func (this_ *Service) CreatePartitions(topic string, count int32) (err error) {
263265
var saramaClient sarama.Client
264266
saramaClient, err = this_.getClient()
265267
if err != nil {
@@ -278,7 +280,7 @@ func (this_ *KafkaService) CreatePartitions(topic string, count int32) (err erro
278280
return
279281
}
280282

281-
func (this_ *KafkaService) CreateTopic(topic string, numPartitions int32, replicationFactor int16) (err error) {
283+
func (this_ *Service) CreateTopic(topic string, numPartitions int32, replicationFactor int16) (err error) {
282284
var saramaClient sarama.Client
283285
saramaClient, err = this_.getClient()
284286
if err != nil {
@@ -306,7 +308,7 @@ func (this_ *KafkaService) CreateTopic(topic string, numPartitions int32, replic
306308
return
307309
}
308310

309-
func (this_ *KafkaService) DeleteTopic(topic string) (err error) {
311+
func (this_ *Service) DeleteTopic(topic string) (err error) {
310312
var saramaClient sarama.Client
311313
saramaClient, err = this_.getClient()
312314
if err != nil {
@@ -325,7 +327,7 @@ func (this_ *KafkaService) DeleteTopic(topic string) (err error) {
325327
return
326328
}
327329

328-
func (this_ *KafkaService) DeleteConsumerGroup(groupId string) (err error) {
330+
func (this_ *Service) DeleteConsumerGroup(groupId string) (err error) {
329331
var saramaClient sarama.Client
330332
saramaClient, err = this_.getClient()
331333
if err != nil {
@@ -344,7 +346,7 @@ func (this_ *KafkaService) DeleteConsumerGroup(groupId string) (err error) {
344346
return
345347
}
346348

347-
func (this_ *KafkaService) DeleteRecords(topic string, partitionOffsets map[int32]int64) (err error) {
349+
func (this_ *Service) DeleteRecords(topic string, partitionOffsets map[int32]int64) (err error) {
348350
var saramaClient sarama.Client
349351
saramaClient, err = this_.getClient()
350352
if err != nil {
@@ -364,7 +366,7 @@ func (this_ *KafkaService) DeleteRecords(topic string, partitionOffsets map[int3
364366
}
365367

366368
// NewSyncProducer 创建生产者
367-
func (this_ *KafkaService) NewSyncProducer() (syncProducer sarama.SyncProducer, err error) {
369+
func (this_ *Service) NewSyncProducer() (syncProducer sarama.SyncProducer, err error) {
368370

369371
config := sarama.NewConfig()
370372
config.Producer.Return.Successes = true
@@ -380,7 +382,7 @@ func (this_ *KafkaService) NewSyncProducer() (syncProducer sarama.SyncProducer,
380382
if this_.CertPath != "" {
381383
certPool := x509.NewCertPool()
382384
var pemCerts []byte
383-
pemCerts, err = ioutil.ReadFile(this_.CertPath)
385+
pemCerts, err = util.ReadFile(this_.CertPath)
384386
if err != nil {
385387
return
386388
}
@@ -410,7 +412,7 @@ type ProducerMessage struct {
410412
}
411413

412414
// Push 推送消息到kafka
413-
func (this_ *KafkaService) Push(msg *Message) (err error) {
415+
func (this_ *Service) Push(msg *Message) (err error) {
414416
producerMessage, err := MessageToProducerMessage(msg)
415417
if err != nil {
416418
return

‎mod_test.go

+65
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
package main
2+
3+
import (
4+
"errors"
5+
"fmt"
6+
"github.com/team-ide/go-tool/util"
7+
"os"
8+
"strings"
9+
"testing"
10+
)
11+
12+
func TestMod(t *testing.T) {
13+
var modPath = `D:\Code\linkdood\vrv-job\job-engine\go.mod`
14+
var modDir = os.Getenv("GOPATH")
15+
modDir = util.FormatPath(modDir)
16+
modDir = modDir + `/pkg/mod`
17+
err := outModTree(0, modPath, modDir)
18+
if err != nil {
19+
panic(err)
20+
}
21+
}
22+
23+
func outModTree(leven int, modPath string, modDir string) (err error) {
24+
lines, err := util.ReadLine(modPath)
25+
if err != nil {
26+
err = errors.New("outModTree ReadLine mod path [" + modPath + "] error:" + err.Error())
27+
return
28+
}
29+
var isInRequire = false
30+
for _, line := range lines {
31+
line = strings.TrimSpace(line)
32+
if isInRequire {
33+
if strings.HasPrefix(line, ")") {
34+
isInRequire = false
35+
break
36+
}
37+
} else {
38+
if strings.HasPrefix(line, "require (") {
39+
isInRequire = true
40+
continue
41+
}
42+
}
43+
if !isInRequire {
44+
continue
45+
}
46+
ss := strings.Split(line, " ")
47+
if len(ss) != 2 {
48+
continue
49+
}
50+
51+
for i := 0; i < leven; i++ {
52+
fmt.Printf("\t")
53+
}
54+
fmt.Printf("%s@%s\n", ss[0], ss[1])
55+
subModPath := modDir + "/" + ss[0] + "@" + ss[1] + "/go.mod"
56+
exi, _ := util.PathExists(subModPath)
57+
if exi {
58+
err = outModTree(leven+1, subModPath, modDir)
59+
if err != nil {
60+
return
61+
}
62+
}
63+
}
64+
return
65+
}

‎package.json

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
11
{
2-
"version": "0.0.4"
2+
"version": "0.0.5"
33
}

‎redis/iface.go

+4
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
package redis
2+
3+
type IService interface {
4+
}

‎redis/service.go

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
package redis

‎util/file.go

+1
Original file line numberDiff line numberDiff line change
@@ -193,6 +193,7 @@ func ReadLine(filename string) (lines []string, err error) {
193193
line = strings.TrimSpace(line)
194194
if err != nil {
195195
if err == io.EOF { //读取结束，会报EOF
196+
err = nil
196197
return
197198
}
198199
return nil, err

‎zookeeper/iface.go

+4
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
package zookeeper
2+
3+
type IService interface {
4+
}

0 commit comments

Comments
 (0)