io.Writer and io.Reader pubsub. Also, CLI tool for watermill

dev/cli/Makefile

@@ -0,0 +1,2 @@
+watermill-cli: 
+	go build -o watermill-cli main.go

dev/cli/README.md

@@ -0,0 +1,62 @@
+# CLI Wrapper for Watermill
+
+This is a CLI wrapper for Watermill. It has two basic functionalities: producing and consuming messages on the following
+Pub/Subs:
+
+1. Kafka
+2. Google Cloud Pub/Sub
+3. RabbitMQ
+
+## Consume mode
+
+In consume mode, the tool subscribes to a topic/queue/subscription (delete as appropriate) and prints the messages
+in a simplified format to the standard output:
+
+```
+[yyyy-mm-dd hh:mm:ss.ssssssss] topic: message payload
+```
+
+Other outputs, for example ones that preserve UUIDs or metadata, are easily attainable by modification
+of the marshaling function of the `io.Publisher` of the `consumeCmd`.
+
+## Produce mode
+
+In produce mode, subsequent lines of data from the standard input are transformed into messages outgoing to the requested
+provider's topic/exchange. 
+
+The message's payload is set to the line from stdin, the UUID is auto-generated and the metadata is empty.
+
+Similarly, the contents of the message could be parsed differently from stdin, by modification
+of the unmarshaling function of the `io.Subscriber` of the `produceCmd`.
+
+## Usage
+
+The basic syntax of the tool is:
+
+```bash
+watermill-cli <provider> <command>
+```
+
+with the appropriate flags regulating the specific behaviour of each command.
+
+`command` is usually one of `produce` or `consume`, but some providers may handle additional commands
+that are specific for them.
+
+The flags are context-specific, so the best way to find out about them is to use the `-h` flag and study 
+which flags are available/required for the specific context and act accordingly.
+
+## Additional functionalities
+
+### Google Cloud Pub/Sub
+
+#### Adding/Removing subscriptions
+
+The CLI tools allows for creating/removing subscriptions for Google Cloud Pub/Sub.
+
+```bash
+watermill-cli googlecloud subscription add -t <topic> <subscription_id>
+
+watermill-cli googlecloud subscription rm <subscription_id>
+```
+
+Additional flags are available for addSubscription to regulate the subscription's settings.

dev/cli/cmd/amqp.go

@@ -0,0 +1,218 @@
+package cmd
+
+import (
+	"github.com/ThreeDotsLabs/watermill/message/infrastructure/amqp"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// amqpCmd is a mid-level command for working with the AMQP Pub/Sub provider.
+var amqpCmd = &cobra.Command{
+	Use:   "amqp",
+	Short: "Commands for the AMQP Pub/Sub provider",
+	Long: `Consume or produce messages from the AMQP Pub/Sub provider.
+
+For the configuration of consuming/producing of the messages, check the help of the relevant command.`,
+	PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
+		err := rootCmd.PersistentPreRunE(cmd, args)
+		if err != nil {
+			return err
+		}
+
+		logger.Debug("Using AMQP Pub/Sub", nil)
+
+		if cmd.Use == "consume" {
+			// amqp is special
+			topic = viper.GetString("amqp.consume.queue")
+			consumer, err = amqp.NewSubscriber(amqpConsumerConfig(), logger)
+			if err != nil {
+				return err
+			}
+		}
+
+		if cmd.Use == "produce" {
+			// amqp is special
+			topic = viper.GetString("amqp.produce.exchange")
+			producer, err = amqp.NewPublisher(amqpProducerConfig(), logger)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	},
+}
+
+func amqpConsumerConfig() amqp.Config {
+	uri := viper.GetString("amqp.uri")
+	queue := viper.GetString("amqp.consume.queue")
+	exchangeName := viper.GetString("amqp.consume.exchange")
+	exchangeType := viper.GetString("amqp.produce.exchangeType")
+	durable := viper.GetBool("amqp.durable")
+
+	return amqp.Config{
+		Connection: amqp.ConnectionConfig{
+			AmqpURI: uri,
+		},
+		Marshaler: amqp.DefaultMarshaler{},
+		Queue: amqp.QueueConfig{
+			GenerateName: func(topic string) string {
+				return queue
+			},
+			Durable: durable,
+		},
+		Consume: amqp.ConsumeConfig{
+			Qos: amqp.QosConfig{
+				PrefetchCount: 1,
+			},
+		},
+
+		Exchange: amqp.ExchangeConfig{
+			GenerateName: func(topic string) string {
+				return exchangeName
+			},
+			Type:    exchangeType,
+			Durable: durable,
+		},
+	}
+}
+
+func amqpProducerConfig() amqp.Config {
+	uri := viper.GetString("amqp.uri")
+	exchangeName := viper.GetString("amqp.produce.exchange")
+	exchangeType := viper.GetString("amqp.produce.exchangeType")
+	routingKey := viper.GetString("amqp.produce.routingKey")
+	durable := viper.GetBool("amqp.durable")
+
+	return amqp.Config{
+		Connection: amqp.ConnectionConfig{
+			AmqpURI: uri,
+		},
+		Marshaler: amqp.DefaultMarshaler{},
+		Exchange: amqp.ExchangeConfig{
+			GenerateName: func(topic string) string {
+				return exchangeName
+			},
+			Type:    exchangeType,
+			Durable: durable,
+		},
+		Publish: amqp.PublishConfig{
+			GenerateRoutingKey: func(topic string) string {
+				return routingKey
+			},
+		},
+	}
+}
+
+func init() {
+	// Here you will define your flags and configuration settings.
+	rootCmd.AddCommand(amqpCmd)
+	configureAmqpCmd()
+	consumeCmd := addConsumeCmd(amqpCmd, false)
+	configureConsumeCmd(consumeCmd)
+	produceCmd := addProduceCmd(amqpCmd, false)
+	configureProduceCmd(produceCmd)
+}
+
+func configureAmqpCmd() {
+	amqpCmd.PersistentFlags().StringP(
+		"uri",
+		"u",
+		"",
+		"The URI to the AMQP instance (required)",
+	)
+	if err := amqpCmd.MarkPersistentFlagRequired("uri"); err != nil {
+		panic(err)
+	}
+	if err := viper.BindPFlag("amqp.uri", amqpCmd.PersistentFlags().Lookup("uri")); err != nil {
+		panic(err)
+	}
+
+	amqpCmd.PersistentFlags().Bool(
+		"durable",
+		true,
+		"If true, the queues and exchanges created automatically (if any) will be durable",
+	)
+	if err := viper.BindPFlag("amqp.durable", amqpCmd.PersistentFlags().Lookup("durable")); err != nil {
+		panic(err)
+	}
+}
+
+func configureConsumeCmd(consumeCmd *cobra.Command) {
+	consumeCmd.PersistentFlags().StringP(
+		"queue",
+		"q",
+		"",
+		"The name of the AMQP queue to consume messages from (required)",
+	)
+	if err := consumeCmd.MarkPersistentFlagRequired("queue"); err != nil {
+		panic(err)
+	}
+	if err := viper.BindPFlag("amqp.consume.queue", consumeCmd.PersistentFlags().Lookup("queue")); err != nil {
+		panic(err)
+	}
+
+	consumeCmd.PersistentFlags().StringP(
+		"exchange",
+		"x",
+		"",
+		"If non-empty, an exchange with this name is created if it didn't exist. Then, the queue is bound to this exchange.",
+	)
+	if err := viper.BindPFlag("amqp.consume.exchange", consumeCmd.PersistentFlags().Lookup("exchange")); err != nil {
+		panic(err)
+	}
+
+	consumeCmd.PersistentFlags().String(
+		"exchangeType",
+		"fanout",
+		"If exchange needs to be created, it will be created with this type. The common types are 'direct', 'fanout', 'topic' and 'headers'.",
+	)
+	if err := consumeCmd.MarkPersistentFlagRequired("exchange"); err != nil {
+		panic(err)
+	}
+
+	if err := viper.BindPFlag("amqp.produce.exchangeType", consumeCmd.PersistentFlags().Lookup("exchangeType")); err != nil {
+		panic(err)
+	}
+}
+
+func configureProduceCmd(produceCmd *cobra.Command) {
+	produceCmd.PersistentFlags().StringP(
+		"exchange",
+		"x",
+		"",
+		"The name of the AMQP exchange to produce messages to (required)",
+	)
+	if err := produceCmd.MarkPersistentFlagRequired("exchange"); err != nil {
+		panic(err)
+	}
+	if err := viper.BindPFlag("amqp.produce.exchange", produceCmd.PersistentFlags().Lookup("exchange")); err != nil {
+		panic(err)
+	}
+
+	produceCmd.PersistentFlags().String(
+		"exchangeType",
+		"fanout",
+		"If the exchange did not exist, it will be created with this type. The common types are 'direct', 'fanout', 'topic' and 'headers'.",
+	)
+	if err := produceCmd.MarkPersistentFlagRequired("exchange"); err != nil {
+		panic(err)
+	}
+
+	if err := viper.BindPFlag("amqp.produce.exchangeType", produceCmd.PersistentFlags().Lookup("exchangeType")); err != nil {
+		panic(err)
+	}
+
+	produceCmd.PersistentFlags().StringP(
+		"routingKey",
+		"r",
+		"",
+		"The routing key to use when publishing the message.",
+	)
+	if err := produceCmd.MarkPersistentFlagRequired("routingKey"); err != nil {
+		panic(err)
+	}
+	if err := viper.BindPFlag("amqp.produce.routingKey", produceCmd.PersistentFlags().Lookup("routingKey")); err != nil {
+		panic(err)
+	}
+}

dev/cli/cmd/consume.go

@@ -0,0 +1,82 @@
+package cmd
+
+import (
+	"os"
+	"time"
+
+	"github.com/ThreeDotsLabs/watermill/message/infrastructure/io"
+	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
+	"github.com/ThreeDotsLabs/watermill/message/router/plugin"
+	"github.com/pkg/errors"
+	"github.com/spf13/viper"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+
+	"github.com/spf13/cobra"
+)
+
+// consumer is initialized by parent command to the pub/sub provider of choice.
+var consumer message.Subscriber
+
+func addConsumeCmd(parent *cobra.Command, addTopicFlag bool) *cobra.Command {
+	parentName := parent.Use
+	cmd := &cobra.Command{
+		Use:   "consume",
+		Short: "Consume messages from a pub/sub and print them to stdout",
+		Long: `Consume messages from the pub/sub of your choice and print them on the standard output.
+
+For the configuration of particular pub/sub providers, see the help for the provider commands.`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			if topic == "" {
+				topic = viper.GetString(parentName + ".consume.topic")
+			}
+			router, err := message.NewRouter(
+				message.RouterConfig{
+					CloseTimeout: 5 * time.Second,
+				},
+				logger,
+			)
+			if err != nil {
+				return errors.Wrap(err, "could not create router")
+			}
+
+			router.AddMiddleware(middleware.InstantAck)
+			router.AddPlugin(plugin.SignalsHandler)
+
+			out, err := io.NewPublisher(os.Stdout, io.PublisherConfig{
+				MarshalFunc: io.PrettyPayloadMarshalFunc,
+			})
+			if err != nil {
+				return errors.Wrap(err, "could not create console producer")
+			}
+
+			router.AddHandler(
+				"dump_to_stdout",
+				topic,
+				consumer,
+				"",
+				out,
+				func(msg *message.Message) ([]*message.Message, error) {
+					// just forward the message to stdout
+					return message.Messages{msg}, nil
+				},
+			)
+
+			return router.Run()
+		},
+	}
+
+	if addTopicFlag {
+		cmd.Flags().StringP("topic", "t", "", "The topic to consume messages from (required)")
+		err := cmd.MarkFlagRequired("topic")
+		if err != nil {
+			panic(err)
+		}
+		if err = viper.BindPFlag(parentName+".consume.topic", cmd.Flags().Lookup("topic")); err != nil {
+			panic(err)
+		}
+	}
+
+	parent.AddCommand(cmd)
+	return cmd
+}