package main

// domain model: bank_account.go

type BankAccount struct {
    Id      string
    Name    string
    Balance int
}

//定义下列函数：

//1. FetchAccount(id) 从Redis读取账户实例信息
//2. updateAccount(id, data) 更新指定账户信息
//3. ToAccount(map) 将从Redis读到的账户信息转换为模型数据，return *BankAccount object.

// main.go
import (
    "github.com/go-redis/redis" // Redis通讯库：go-redis
)

var (
    Redis = initRedis()
)

func initRedis() *redis.Client {
    redisUrl := os.Getenv("REDIS_URL")

    if redisUrl == "" {
        redisUrl = "127.0.0.1:6379"
    }

    return redis.NewClient(&redis.Options{
        Addr:     redisUrl,
        Password: "",
        DB:       0,
    })
}

package main
//kafka.go
import (
    "encoding/json"
    "fmt"
    "github.com/Shopify/sarama" //Kafka通讯库：Sarama
    "os"
)

var (
    brokers = []string{"127.0.0.1:9092"}
    topic   = "go-microservice-transactions"
    topics  = []string{topic}
)

func newKafkaConfiguration() *sarama.Config {
    conf := sarama.NewConfig()
    conf.Producer.RequiredAcks = sarama.WaitForAll
    conf.Producer.Return.Successes = true
    conf.ChannelBufferSize = 1
    conf.Version = sarama.V0_10_1_0
    return conf
}

func newKafkaSyncProducer() sarama.SyncProducer {
    kafka, err := sarama.NewSyncProducer(brokers, newKafkaConfiguration())

    if err != nil {
        fmt.Printf("Kafka error: %s\n", err)
        os.Exit(-1)
    }
    return kafka
}

func newKafkaConsumer() sarama.Consumer {
    consumer, err := sarama.NewConsumer(brokers, newKafkaConfiguration())

    if err != nil {
        fmt.Printf("Kafka error: %s\n", err)
        os.Exit(-1)
    }

    return consumer
}

package main
//消息生产者 producer.go
import (
    "bufio"
    "fmt"
    "os"
    "strconv"
    "strings"
)

func mainProducer() {
    var err error
    reader := bufio.NewReader(os.Stdin)
    kafka := newKafkaSyncProducer()

    for {
        fmt.Print("-> ")
        text, _ := reader.ReadString('\n')
        text = strings.Replace(text, "\n", "", -1)
        args := strings.Split(text, "###")
        cmd := args[0]

        switch cmd {
        case "create":
            if len(args) == 2 {
                accName := args[1]
                event := NewCreateAccountEvent(accName)
                sendMsg(kafka, event)
            } else {
                fmt.Println("Only specify create###Account Name")
            }
        default:
            fmt.Printf("Unknown command %s, only: create, deposit, withdraw, transfer\n", cmd)
        }

        if err != nil {
            fmt.Printf("Error: %s\n", err)
            err = nil
        }
    }
}

// kafka.go
// 增加发送消息的方法

func sendMsg(kafka sarama.SyncProducer, event interface{}) error {
    json, err := json.Marshal(event)

    if err != nil {
        return err
    }

    msgLog := &sarama.ProducerMessage{
        Topic: topic,
        Value: sarama.StringEncoder(string(json)),
    }

    partition, offset, err := kafka.SendMessage(msgLog)
    if err != nil {
        fmt.Printf("Kafka error: %s\n", err)
    }

    fmt.Printf("Message: %+v\n", event)
    fmt.Printf("Message is stored in partition %d, offset %d\n",
        partition, offset)

    return nil
}

package main
//启动入口，main.go

func main() {
    mainProducer()
}

package main
//processor.go
import (
  "errors"
)
func (e CreateEvent) Process() (*BankAccount, error) {
    return updateAccount(e.AccId, map[string]interface{}{
        "Id":      e.AccId,
        "Name":    e.AccName,
        "Balance": "0",
    })
}

func (e InvalidEvent) Process() error {
  return nil
}

func (e AcceptEvent) Process() error {
  return nil
}
// other Process() codes ...

package main

//consumer.go

func mainConsumer(partition int32) {
    kafka := newKafkaConsumer()
    defer kafka.Close()
    //注：开发环境中我们使用sarama.OffsetOldest，Kafka将从创建以来第一条消息开始发送。
    //在生产环境中切换为sarama.OffsetNewest,只会将最新生成的消息发送给我们。
    consumer, err := kafka.ConsumePartition(topic, partition, sarama.OffsetOldest)
    if err != nil {
        fmt.Printf("Kafka error: %s\n", err)
        os.Exit(-1)
    }

    go consumeEvents(consumer)

    fmt.Println("Press [enter] to exit consumer\n")
    bufio.NewReader(os.Stdin).ReadString('\n')
    fmt.Println("Terminating...")
}

func consumeEvents(consumer sarama.PartitionConsumer) {
  var msgVal []byte
  var log interface{}
  var logMap map[string]interface{}
  var bankAccount *BankAccount
  var err error

  for {
    //goruntine exec
      select {
          // blocking <- channel operator
          case err := <-consumer.Errors():
              fmt.Printf("Kafka error: %s\n", err)
          case msg := <-consumer.Messages():
              msgVal = msg.Value
          //
          if err = json.Unmarshal(msgVal, &log); err != nil {
                fmt.Printf("Failed parsing: %s", err)
          } else {
                logMap = log.(map[string]interface{})
                logType := logMap["Type"]
                fmt.Printf("Processing %s:\n%s\n", logMap["Type"], string(msgVal))

                switch logType {
                case "CreateEvent":
                  event := new(CreateEvent)
                  if err = json.Unmarshal(msgVal, &event); err == nil {
                    bankAccount, err = event.Process()
                  }
                default:
                  fmt.Println("Unknown command: ", logType)
                }

                if err != nil {
                  fmt.Printf("Error processing: %s\n", err)
                  } else {
                    fmt.Printf("%+v\n\n", *bankAccount)
                  }
            }

        }
    }
}  

package main

//main.go
//支持producer和consumer启动模式

import (
    "flag"
    ...
)

func main() {
    act := flag.String("act", "producer", "Either: producer or consumer")
    partition := flag.String("partition", "0",
        "Partition which the consumer program will be subscribing")

    flag.Parse()

    fmt.Printf("Welcome to go-microservice : %s\n\n", *act)

    switch *act {
    case "producer":
        mainProducer()
    case "consumer":
        if part32int, err := strconv.ParseInt(*partition, 10, 32); err == nil {
            mainConsumer(int32(part32int))
        }
    }
}

//修改mainConsumer方法使用sarama-cluster library连接Kafka
config := cluster.NewConfig()
config.Consumer.Offsets.Initial = sarama.OffsetNewest
consumer, err := cluster.NewConsumer(brokers, "go-microservice-consumer", topics, config)

//topics定义
var (
    topics  = []string{topic}
)

//调整consumeEvents()
case err, more := <-consumer.Errors():
    if more {
        fmt.Printf("Kafka error: %s\n", err)
    }

//consumer.Messages() : MarkOffset
//consumer.go
//func mainConsumer(partition int32)

consumer.MarkOffset(msg, "") //增加的行

msgVal = msg.Value

//运行多个consumer实例
$ ./go-microservice --act=consumer
$ ./go-microservice --act=consumer
$ ./go-microservice --act=consumer

$ //
$ go get -u github.com/kardianos/govendor
$ cd $PROJECT_PATH
$ govendor init
$ govendor add +external
$

$ go get github.com/onsi/ginkgo/ginkgo
$ go get github.com/onsi/gomega
$ ginkgo bootstrap
Generating ginkgo test suite bootstrap for main in:
	go_microservice_suite_test.go

package main_test
//go_microservice_suite_test.go,单元测试类
import (
    "github.com/onsi/ginkgo"
    "github.com/onsi/gomega"
)

var _ = Describe("Event", func() {
    Describe("NewCreateAccountEvent", func() {
        It("can create a create account event", func() {
            name := "John Smith"

            event := NewCreateAccountEvent(name)

            Expect(event.AccName).To(Equal(name))
            Expect(event.AccId).NotTo(BeNil())
            Expect(event.Type).To(Equal("CreateEvent"))
        })
    })
})

//install our dependencies
RUN go get -u github.com/kardianos/govendor
RUN go get github.com/onsi/ginkgo/ginkgo
RUN go get github.com/onsi/gomega

//将整个目录拷贝到容器
ADD . /go/src/go-microservice

//检查工作目录
WORKDIR /go/src/go-microservice

//安装依赖项
RUN govendor sync

//测试
$ docker build -t go-microservice .
$ docker run -i -t go-microservice /bin/bash
$ ginkgo
.......................
.......Failed..........