package main

import (
    "flag"
    "fmt"
    "log"
    "os"
    "os/signal"
    "strconv"
    "strings"
    "sync"
    "syscall"

    "github.com/Shopify/sarama"
    "github.com/Shopify/sarama/tools/tls"
)

// 从命令行终端获取参数
var (
    brokerList    = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The comma separated list of brokers in the Kafka cluster")
    topic         = flag.String("topic", "", "REQUIRED: the topic to consume")
    partitions    = flag.String("partitions", "all", "The partitions to consume, can be 'all' or comma-separated numbers")
    offset        = flag.String("offset", "newest", "The offset to start with. Can be `oldest`, `newest`")
    verbose       = flag.Bool("verbose", false, "Whether to turn on sarama logging")
    tlsEnabled    = flag.Bool("tls-enabled", false, "Whether or not to use TLS when connecting to the broker  (defaults to false)")
    tlsSkipVerify = flag.Bool("tls-skip-verify", false, "Whether skip TLS server cert verification")
    tlsClientCert = flag.String("tls-client-cert", "", "Client cert for client authentication (use with -tls-enabled and -tls-client-key)")
    tlsClientKey  = flag.String("tls-client-key", "", "Client key for client authentication (use with tls-enabled and -tls-client-cert)")
    saslEnabled   = flag.Bool("sasl-enabled", false, "Whether or not to use SASL authentication when connecting to the broker (defaults to false)")
    saslUser      = flag.String("sasl-user", "", "The username used for SASL (use with -sasl-enable)")
    saslPassword  = flag.String("sasl-password", "", "The password used for SASL (use with -sasl-enable)")

    bufferSize = flag.Int("buffer-size", 256, "The buffer size of the message channel.")

    logger = log.New(os.Stderr, "", log.LstdFlags)
)

func main() {
    flag.Parse()

    if *brokerList == "" {
        printUsageErrorAndExit("You have to provide -brokers as a comma-separated list, or set the KAFKA_PEERS environment variable.")
    }

    if *topic == "" {
        printUsageErrorAndExit("-topic is required")
    }

    if *verbose {
        sarama.Logger = logger
    }

    // 初始化读取位置；
    var initialOffset int64
    switch *offset {
    case "oldest":
        initialOffset = sarama.OffsetOldest
    case "newest":
        initialOffset = sarama.OffsetNewest
    default:
        printUsageErrorAndExit("-offset should be `oldest` or `newest`")
    }

    // 初始化kafka配置参数
    config := sarama.NewConfig()
    // 配置kafka版本
    config.Version = sarama.V0_10_1_0
    // 配置TLS相关
    if *tlsEnabled {
        tlsConfig, err := tls.NewConfig(*tlsClientCert, *tlsClientKey)
        if err != nil {
            printErrorAndExit(69, "Failed to create TLS config: %s", err)
        }

        config.Net.TLS.Enable = true
        config.Net.TLS.Config = tlsConfig
        config.Net.TLS.Config.InsecureSkipVerify = *tlsSkipVerify
    }

    // 配置用户名密码认证
    if *saslEnabled {
        config.Net.SASL.Enable = true
        config.Net.SASL.User = *saslUser
        config.Net.SASL.Password = *saslPassword
    }

    // 实例化consumer
    c, err := sarama.NewConsumer(strings.Split(*brokerList, ","), config)
    if err != nil {
        printErrorAndExit(69, "Failed to start consumer: %s", err)
    }

    // 获取partition
    partitionList, err := getPartitions(c)
    if err != nil {
        printErrorAndExit(69, "Failed to get the list of partitions: %s", err)
    }

    // 定义channel
    var (
        messages = make(chan *sarama.ConsumerMessage, *bufferSize)
        closing  = make(chan struct{})
        wg       sync.WaitGroup
    )

    // goroutin 等待系统中断信号
    go func() {
        signals := make(chan os.Signal, 1)
        signal.Notify(signals, syscall.SIGTERM, os.Interrupt)
        <-signals
        logger.Println("Initiating shutdown of consumer...")
        close(closing)
    }()

    // 根据partitionlist启动多个goroutin去消费
    for _, partition := range partitionList {
        pc, err := c.ConsumePartition(*topic, partition, initialOffset)
        if err != nil {
            printErrorAndExit(69, "Failed to start consumer for partition %d: %s", partition, err)
        }

        go func(pc sarama.PartitionConsumer) {
            <-closing
            pc.AsyncClose()
        }(pc)

        wg.Add(1)
        go func(pc sarama.PartitionConsumer) {
            defer wg.Done()
            for message := range pc.Messages() {
                messages <- message
            }
        }(pc)
    }

    go func() {
        for msg := range messages {
            fmt.Printf("Partition:\t%d\n", msg.Partition)
            fmt.Printf("Offset:\t%d\n", msg.Offset)
            fmt.Printf("Key:\t%s\n", string(msg.Key))
            fmt.Printf("Value:\t%s\n", string(msg.Value))
            fmt.Println()
        }
    }()

    wg.Wait()
    logger.Println("Done consuming topic", *topic)
    close(messages)

    if err := c.Close(); err != nil {
        logger.Println("Failed to close consumer: ", err)
    }
}

func getPartitions(c sarama.Consumer) ([]int32, error) {
    if *partitions == "all" {
        return c.Partitions(*topic)
    }

    tmp := strings.Split(*partitions, ",")
    var pList []int32
    for i := range tmp {
        val, err := strconv.ParseInt(tmp[i], 10, 32)
        if err != nil {
            return nil, err
        }
        pList = append(pList, int32(val))
    }

    return pList, nil
}

func printErrorAndExit(code int, format string, values ...interface{}) {
    fmt.Fprintf(os.Stderr, "ERROR: %s\n", fmt.Sprintf(format, values...))
    fmt.Fprintln(os.Stderr)
    os.Exit(code)
}

func printUsageErrorAndExit(format string, values ...interface{}) {
    fmt.Fprintf(os.Stderr, "ERROR: %s\n", fmt.Sprintf(format, values...))
    fmt.Fprintln(os.Stderr)
    fmt.Fprintln(os.Stderr, "Available command line options:")
    flag.PrintDefaults()
    os.Exit(64)
}

package main

import (
    "flag"
    "fmt"
    "log"
    "os"
    "os/signal"
    "strconv"
    "strings"
    "sync"
    "syscall"

    "github.com/Shopify/sarama"
    "github.com/Shopify/sarama/tools/tls"
)

// 从命令行终端获取参数
var (
    brokerList    = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The comma separated list of brokers in the Kafka cluster")
    topic         = flag.String("topic", "", "REQUIRED: the topic to consume")
    partitions    = flag.String("partitions", "all", "The partitions to consume, can be 'all' or comma-separated numbers")
    offset        = flag.String("offset", "newest", "The offset to start with. Can be `oldest`, `newest`")
    verbose       = flag.Bool("verbose", false, "Whether to turn on sarama logging")
    tlsEnabled    = flag.Bool("tls-enabled", false, "Whether or not to use TLS when connecting to the broker  (defaults to false)")
    tlsSkipVerify = flag.Bool("tls-skip-verify", false, "Whether skip TLS server cert verification")
    tlsClientCert = flag.String("tls-client-cert", "", "Client cert for client authentication (use with -tls-enabled and -tls-client-key)")
    tlsClientKey  = flag.String("tls-client-key", "", "Client key for client authentication (use with tls-enabled and -tls-client-cert)")
    saslEnabled   = flag.Bool("sasl-enabled", false, "Whether or not to use SASL authentication when connecting to the broker (defaults to false)")
    saslUser      = flag.String("sasl-user", "", "The username used for SASL (use with -sasl-enable)")
    saslPassword  = flag.String("sasl-password", "", "The password used for SASL (use with -sasl-enable)")

    bufferSize = flag.Int("buffer-size", 256, "The buffer size of the message channel.")

    logger = log.New(os.Stderr, "", log.LstdFlags)
)

func main() {
    flag.Parse()

    if *brokerList == "" {
        printUsageErrorAndExit("You have to provide -brokers as a comma-separated list, or set the KAFKA_PEERS environment variable.")
    }

    if *topic == "" {
        printUsageErrorAndExit("-topic is required")
    }

    if *verbose {
        sarama.Logger = logger
    }

    // 初始化读取位置；
    var initialOffset int64
    switch *offset {
    case "oldest":
        initialOffset = sarama.OffsetOldest
    case "newest":
        initialOffset = sarama.OffsetNewest
    default:
        printUsageErrorAndExit("-offset should be `oldest` or `newest`")
    }

    // 初始化kafka配置参数
    config := sarama.NewConfig()
    // 配置kafka版本
    config.Version = sarama.V0_10_1_0
    // 配置TLS相关
    if *tlsEnabled {
        tlsConfig, err := tls.NewConfig(*tlsClientCert, *tlsClientKey)
        if err != nil {
            printErrorAndExit(69, "Failed to create TLS config: %s", err)
        }

        config.Net.TLS.Enable = true
        config.Net.TLS.Config = tlsConfig
        config.Net.TLS.Config.InsecureSkipVerify = *tlsSkipVerify
    }

    // 配置用户名密码认证
    if *saslEnabled {
        config.Net.SASL.Enable = true
        config.Net.SASL.User = *saslUser
        config.Net.SASL.Password = *saslPassword
    }

    // 实例化consumer
    c, err := sarama.NewConsumer(strings.Split(*brokerList, ","), config)
    if err != nil {
        printErrorAndExit(69, "Failed to start consumer: %s", err)
    }

    // 获取partition
    partitionList, err := getPartitions(c)
    if err != nil {
        printErrorAndExit(69, "Failed to get the list of partitions: %s", err)
    }

    // 定义channel
    var (
        messages = make(chan *sarama.ConsumerMessage, *bufferSize)
        closing  = make(chan struct{})
        wg       sync.WaitGroup
    )

    // goroutin 等待系统中断信号
    go func() {
        signals := make(chan os.Signal, 1)
        signal.Notify(signals, syscall.SIGTERM, os.Interrupt)
        <-signals
        logger.Println("Initiating shutdown of consumer...")
        close(closing)
    }()

    // 根据partitionlist启动多个goroutin去消费
    for _, partition := range partitionList {
        pc, err := c.ConsumePartition(*topic, partition, initialOffset)
        if err != nil {
            printErrorAndExit(69, "Failed to start consumer for partition %d: %s", partition, err)
        }

        go func(pc sarama.PartitionConsumer) {
            <-closing
            pc.AsyncClose()
        }(pc)

        wg.Add(1)
        go func(pc sarama.PartitionConsumer) {
            defer wg.Done()
            for message := range pc.Messages() {
                messages <- message
            }
        }(pc)
    }

    go func() {
        for msg := range messages {
            fmt.Printf("Partition:\t%d\n", msg.Partition)
            fmt.Printf("Offset:\t%d\n", msg.Offset)
            fmt.Printf("Key:\t%s\n", string(msg.Key))
            fmt.Printf("Value:\t%s\n", string(msg.Value))
            fmt.Println()
        }
    }()

    wg.Wait()
    logger.Println("Done consuming topic", *topic)
    close(messages)

    if err := c.Close(); err != nil {
        logger.Println("Failed to close consumer: ", err)
    }
}

func getPartitions(c sarama.Consumer) ([]int32, error) {
    if *partitions == "all" {
        return c.Partitions(*topic)
    }

    tmp := strings.Split(*partitions, ",")
    var pList []int32
    for i := range tmp {
        val, err := strconv.ParseInt(tmp[i], 10, 32)
        if err != nil {
            return nil, err
        }
        pList = append(pList, int32(val))
    }

    return pList, nil
}

func printErrorAndExit(code int, format string, values ...interface{}) {
    fmt.Fprintf(os.Stderr, "ERROR: %s\n", fmt.Sprintf(format, values...))
    fmt.Fprintln(os.Stderr)
    os.Exit(code)
}

func printUsageErrorAndExit(format string, values ...interface{}) {
    fmt.Fprintf(os.Stderr, "ERROR: %s\n", fmt.Sprintf(format, values...))
    fmt.Fprintln(os.Stderr)
    fmt.Fprintln(os.Stderr, "Available command line options:")
    flag.PrintDefaults()
    os.Exit(64)
}