diff --git a/controller/teardown.go b/controller/close.go similarity index 87% rename from controller/teardown.go rename to controller/close.go index b39931d..039961a 100644 --- a/controller/teardown.go +++ b/controller/close.go @@ -6,7 +6,7 @@ import ( "github.com/gin-gonic/gin" ) -func Teardown(c *gin.Context) { +func Close(c *gin.Context) { // Close connection err := client.Teardown(c) diff --git a/controller/open.go b/controller/open.go index 1e933d5..836edf1 100644 --- a/controller/open.go +++ b/controller/open.go @@ -1,6 +1,7 @@ package controller import ( + "log" "net/http" "elasticstream/config" @@ -13,10 +14,12 @@ import ( var client *elastic.Client func Open(c *gin.Context) { + log.Println(">>>> controller.Open()") + defer log.Println("<<<< controller.Open()") client = elastic.NewClient() - err := client.Configure(c, config.Config{ + err := client.Configure(c, &config.Config{ Host: "http://test.urantiacloud.com:9200", Indexes: []string{"index-a", "index-b", "index-c"}, BatchSize: 100, @@ -30,6 +33,8 @@ func Open(c *gin.Context) { // get position from boltdb var positions []source.Position + log.Printf("client: %#v\n", client) + err = client.Open(c, positions) if err != nil { c.Status(http.StatusInternalServerError) diff --git a/main.go b/main.go index e5456ba..71770d7 100644 --- a/main.go +++ b/main.go @@ -1,6 +1,8 @@ package main import ( + "log" + "elasticstream/controller" "github.com/gin-gonic/gin" @@ -8,12 +10,14 @@ import ( func main() { + log.SetFlags(log.LstdFlags | log.Lshortfile) + r := gin.Default() r.GET("/open", controller.Open) r.GET("/read", controller.Read) r.GET("/ack", controller.Ack) - r.DELETE("/teardown", controller.Teardown) + r.GET("/close", controller.Close) r.Run(":8080") } diff --git a/source/elastic/configure.go b/source/elastic/configure.go index 03e065c..64d4b29 100644 --- a/source/elastic/configure.go +++ b/source/elastic/configure.go @@ -2,10 +2,17 @@ package elastic import ( "context" + "fmt" "elasticstream/config" ) -func (c *Client) Configure(ctx context.Context, cfg config.Config) error { +func (c *Client) Configure(ctx context.Context, cfg *config.Config) error { + + if c == nil || c.ch == nil { + return fmt.Errorf("error source not opened for reading") + } + + c.cfg = cfg return nil } diff --git a/source/elastic/db.go b/source/elastic/db.go deleted file mode 100644 index 252e851..0000000 --- a/source/elastic/db.go +++ /dev/null @@ -1,40 +0,0 @@ -package elastic - -import ( - "fmt" - "log" - "strconv" - - "github.com/boltdb/bolt" -) - -func getOffset(db *bolt.DB, index string) (int, error) { - offset := 0 - if err := db.Update(func(tx *bolt.Tx) error { - b, err := tx.CreateBucketIfNotExists([]byte("position")) - if err != nil { - return err - } - v := b.Get([]byte(index)) - offset, _ = strconv.Atoi(string(v)) - return nil - }); err != nil { - return offset, err - } - return offset, nil -} - -func setOffset(db *bolt.DB, index string, offset int) error { - // log.Printf("setOffset() index=%s offset=%d", index, offset) - if err := db.Update(func(tx *bolt.Tx) error { - b := tx.Bucket([]byte("position")) - - if err := b.Put([]byte(index), []byte(fmt.Sprintf("%d", offset))); err != nil { - return err - } - return nil - }); err != nil { - log.Fatal(err) - } - return nil -} diff --git a/source/elastic/open.go b/source/elastic/open.go index 61ad0ef..2d07439 100644 --- a/source/elastic/open.go +++ b/source/elastic/open.go @@ -2,6 +2,7 @@ package elastic import ( "context" + "log" "elasticstream/opencdc" "elasticstream/source" @@ -10,6 +11,8 @@ import ( ) func (c *Client) Open(ctx context.Context, positions []source.Position) error { + log.Println(">>>> elastic.Open()") + defer log.Println("<<<< elastic.Open()") // Open a connection with ElasticSearch cfg := elasticsearch.Config{ diff --git a/source/elastic/read.go b/source/elastic/read.go index 3312f0b..3e4f6af 100644 --- a/source/elastic/read.go +++ b/source/elastic/read.go @@ -3,11 +3,18 @@ package elastic import ( "context" "fmt" + "log" "elasticstream/opencdc" ) func (c *Client) Read(context.Context) (*opencdc.Data, error) { + log.Println(">>>> elastic.Read()") + defer log.Println("<<<< elastic.Read()") + + if c == nil || c.ch == nil { + return nil, fmt.Errorf("error source not opened for reading") + } data, ok := <-c.ch if !ok { @@ -26,10 +33,5 @@ func (c *Client) Read(context.Context) (*opencdc.Data, error) { c.offsets[index] = offset + 1 - err := setOffset(c.db, index, c.offsets[index]) - if err != nil { - return nil, err - } - return &data, nil } diff --git a/source/elastic/teardown.go b/source/elastic/teardown.go index 94389c0..3a41497 100644 --- a/source/elastic/teardown.go +++ b/source/elastic/teardown.go @@ -2,9 +2,14 @@ package elastic import ( "context" + "fmt" ) // close the client func (c *Client) Teardown(ctx context.Context) error { + if c == nil || c.ch == nil { + return fmt.Errorf("error source not opened for reading") + } + return nil }