feat: added functionality for replication create command (#289)

* feat: added functionality for replication create command

* refactor: changed replication create flag usage text and eliminated extra conditionals

* refactor: made changes to usage text and output table based on code review

* chore: fixed link to max queue size default value

clients/replication/replication.go

@@ -0,0 +1,106 @@
+package replication
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/influxdata/influx-cli/v2/api"
+	"github.com/influxdata/influx-cli/v2/clients"
+)
+
+type Client struct {
+	clients.CLI
+	api.ReplicationsApi
+	api.OrganizationsApi
+}
+
+type CreateParams struct {
+	Name           string
+	Description    string
+	OrgID          string
+	OrgName        string
+	RemoteID       string
+	LocalBucketID  string
+	RemoteBucketID string
+	MaxQueueSize   int64
+}
+
+func (c Client) Create(ctx context.Context, params *CreateParams) error {
+	orgID, err := c.GetOrgId(ctx, params.OrgID, params.OrgName, c.OrganizationsApi)
+	if err != nil {
+		return err
+	}
+
+	// set up a struct with required params
+	body := api.ReplicationCreationRequest{
+		Name:              params.Name,
+		OrgID:             orgID,
+		RemoteID:          params.RemoteID,
+		LocalBucketID:     params.LocalBucketID,
+		RemoteBucketID:    params.RemoteBucketID,
+		MaxQueueSizeBytes: params.MaxQueueSize,
+	}
+
+	// set optional params if specified
+	if params.Description != "" {
+		body.Description = &params.Description
+	}
+
+	// send post request
+	res, err := c.PostReplication(ctx).ReplicationCreationRequest(body).Execute()
+	if err != nil {
+		return fmt.Errorf("failed to create replication stream %q: %w", params.Name, err)
+	}
+
+	// print confirmation of new replication stream
+	return c.printReplication(printReplicationOpts{replication: &res})
+}
+
+type printReplicationOpts struct {
+	replication  *api.Replication
+	replications []api.Replication
+	deleted      bool
+}
+
+func (c Client) printReplication(opts printReplicationOpts) error {
+	if c.PrintAsJSON {
+		var v interface{}
+		if opts.replication != nil {
+			v = opts.replication
+		} else {
+			v = opts.replications
+		}
+		return c.PrintJSON(v)
+	}
+
+	headers := []string{"ID", "Name", "Org ID", "Remote ID", "Local Bucket ID", "Remote Bucket ID", "Max Queue Bytes",
+		"Current Queue Bytes", "Latest Status Code"}
+	if opts.deleted {
+		headers = append(headers, "Deleted")
+	}
+
+	if opts.replication != nil {
+		opts.replications = append(opts.replications, *opts.replication)
+	}
+
+	var rows []map[string]interface{}
+	for _, r := range opts.replications {
+		row := map[string]interface{}{
+			"ID":                  r.GetId(),
+			"Name":                r.GetName(),
+			"Org ID":              r.GetOrgID(),
+			"Remote ID":           r.GetRemoteID(),
+			"Local Bucket ID":     r.GetLocalBucketID(),
+			"Remote Bucket ID":    r.GetRemoteBucketID(),
+			"Max Queue Bytes":     r.GetMaxQueueSizeBytes(),
+			"Current Queue Bytes": r.GetCurrentQueueSizeBytes(),
+			"Latest Status Code":  r.GetLatestResponseCode(),
+		}
+		if opts.deleted {
+			row["Deleted"] = true
+		}
+		rows = append(rows, row)
+	}
+
+	return c.PrintTable(headers, rows...)
+}

cmd/influx/replication.go

@@ -3,6 +3,7 @@ package main
 import (
 	"fmt"
 
+	"github.com/influxdata/influx-cli/v2/clients/replication"
 	"github.com/influxdata/influx-cli/v2/pkg/cli/middleware"
 	"github.com/urfave/cli"
 )
@@ -22,13 +23,71 @@ func newReplicationCmd() cli.Command {
 }
 
 func newReplicationCreateCmd() cli.Command {
+	var params replication.CreateParams
 	return cli.Command{
 		Name:   "create",
 		Usage:  "Create a new replication stream",
 		Before: middleware.WithBeforeFns(withCli(), withApi(true), middleware.NoArgs),
-		Flags:  commonFlags(),
-		Action: func(ctx *cli.Context) {
-			fmt.Println("replication create command was called")
+		Flags: append(
+			commonFlags(),
+			&cli.StringFlag{
+				Name:        "name, n",
+				Usage:       "Name for new replication stream",
+				Required:    true,
+				Destination: &params.Name,
+			},
+			&cli.StringFlag{
+				Name:        "description, d",
+				Usage:       "Description for new replication stream",
+				Destination: &params.Description,
+			},
+			&cli.StringFlag{
+				Name:        "org-id",
+				Usage:       "The ID of the local organization",
+				EnvVar:      "INFLUX_ORG_ID",
+				Destination: &params.OrgID,
+			},
+			&cli.StringFlag{
+				Name:        "org, o",
+				Usage:       "The name of the local organization",
+				EnvVar:      "INFLUX_ORG",
+				Destination: &params.OrgName,
+			},
+			&cli.StringFlag{
+				Name:        "remote-id",
+				Usage:       "Remote connection the new replication stream should send data to",
+				Required:    true,
+				Destination: &params.RemoteID,
+			},
+			&cli.StringFlag{
+				Name:        "local-bucket",
+				Usage:       "ID of local bucket data should be replicated from",
+				Required:    true,
+				Destination: &params.LocalBucketID,
+			},
+			&cli.StringFlag{
+				Name:        "remote-bucket",
+				Usage:       "ID of remote bucket data should be replicated to",
+				Required:    true,
+				Destination: &params.RemoteBucketID,
+			},
+			&cli.Int64Flag{
+				Name:        "max-queue-bytes",
+				Usage:       "Max queue size in bytes",
+				Value:       67108860, // source: https://github.com/influxdata/openapi/blob/588064fe68e7dfeebd019695aa805832632cbfb6/src/oss/schemas/ReplicationCreationRequest.yml#L19
+				Destination: &params.MaxQueueSize,
+			},
+		),
+		Action: func(ctx *cli.Context) error {
+			api := getAPI(ctx)
+
+			client := replication.Client{
+				CLI:              getCLI(ctx),
+				ReplicationsApi:  api.ReplicationsApi,
+				OrganizationsApi: api.OrganizationsApi,
+			}
+
+			return client.Create(getContext(ctx), &params)
 		},
 	}
 }