// MySQLOnly is used to launch only a mysqld instance, with the specified db name.
// Use it before Schema option.
// It cannot be used at the same as ProtoTopo.
func MySQLOnly(dbName string) VitessOption {
	return VitessOption{
		beforeRun: func(hdl *Handle) error {
			// the way to pass the dbname for creation in
			// is to provide a topology
			topo := &vttestpb.VTTestTopology{
				Keyspaces: []*vttestpb.Keyspace{
					{
						Name: dbName,
						Shards: []*vttestpb.Shard{
							{
								Name:           "0",
								DbNameOverride: dbName,
							},
						},
					},
				},
			}

			hdl.dbname = dbName
			hdl.cmd.Args = append(hdl.cmd.Args,
				"--mysql_only",
				"--proto_topo", proto.CompactTextString(topo))
			return nil
		},
	}
}

// ProtoTopo is used to pass in the topology as a vttest proto definition.
// See vttest.proto for more information.
// It cannot be used at the same time as MySQLOnly.
func ProtoTopo(topo *vttestpb.VTTestTopology) VitessOption {
	return VitessOption{
		beforeRun: func(hdl *Handle) error {
			hdl.cmd.Args = append(hdl.cmd.Args, "--proto_topo", proto.CompactTextString(topo))
			return nil
		},
	}
}

// tests that a Getter's Get method is only called once with two
// outstanding callers.  This is the proto variant.
func TestGetDupSuppressProto(t *testing.T) {
	once.Do(testSetup)
	// Start two getters. The first should block (waiting reading
	// from stringc) and the second should latch on to the first
	// one.
	resc := make(chan *testpb.TestMessage, 2)
	for i := 0; i < 2; i++ {
		go func() {
			tm := new(testpb.TestMessage)
			if err := protoGroup.Get(dummyCtx, fromChan, ProtoSink(tm)); err != nil {
				tm.Name = proto.String("ERROR:" + err.Error())
			}
			resc <- tm
		}()
	}

	// Wait a bit so both goroutines get merged together via
	// singleflight.
	// TODO(bradfitz): decide whether there are any non-offensive
	// debug/test hooks that could be added to singleflight to
	// make a sleep here unnecessary.
	time.Sleep(250 * time.Millisecond)

	// Unblock the first getter, which should unblock the second
	// as well.
	stringc <- "Fluffy"
	want := &testpb.TestMessage{
		Name: proto.String("ECHO:Fluffy"),
		City: proto.String("SOME-CITY"),
	}
	for i := 0; i < 2; i++ {
		select {
		case v := <-resc:
			if !reflect.DeepEqual(v, want) {
				t.Errorf(" Got: %v\nWant: %v", proto.CompactTextString(v), proto.CompactTextString(want))
			}
		case <-time.After(5 * time.Second):
			t.Errorf("timeout waiting on getter #%d of 2", i+1)
		}
	}
}

// WritePB writes a proto message to the writer.
func WritePB(writer io.Writer, pb proto.Message) error {
	payloadBytes, err := proto.Marshal(pb)
	if err != nil {
		return fmt.Errorf("Failed to marshal proto: %s", proto.CompactTextString(pb))
	}

	payloadSize := [4]byte{}
	binary.BigEndian.PutUint32(payloadSize[:], uint32(len(payloadBytes)))
	if _, err = writer.Write(payloadSize[:]); err != nil {
		return fmt.Errorf("Failed to write 4 bytes for size: %s", err)
	}
	if _, err = writer.Write(payloadBytes); err != nil {
		return fmt.Errorf("Failed to write %d bytes to payload: %s", payloadSize, err)
	}
	return nil
}

func commandGetThrottlerConfiguration(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error {
	server := subFlags.String("server", "", "vtworker or vttablet to connect to")
	if err := subFlags.Parse(args); err != nil {
		return err
	}
	if subFlags.NArg() > 1 {
		return fmt.Errorf("the GetThrottlerConfiguration command accepts only <throttler name> as optional positional parameter")
	}

	var throttlerName string
	if subFlags.NArg() == 1 {
		throttlerName = subFlags.Arg(0)
	}

	// Connect to the server.
	ctx, cancel := context.WithTimeout(ctx, shortTimeout)
	defer cancel()
	client, err := throttlerclient.New(*server)
	if err != nil {
		return fmt.Errorf("error creating a throttler client for server '%v': %v", *server, err)
	}
	defer client.Close()

	configurations, err := client.GetConfiguration(ctx, throttlerName)
	if err != nil {
		return fmt.Errorf("failed to get the throttler configuration from server '%v': %v", *server, err)
	}

	if len(configurations) == 0 {
		wr.Logger().Printf("There are no active throttlers on server '%v'.\n", *server)
		return nil
	}

	table := tablewriter.NewWriter(loggerWriter{wr.Logger()})
	table.SetAutoFormatHeaders(false)
	// The full protobuf text will span more than one terminal line. Do not wrap
	// it to make it easy to copy and paste it.
	table.SetAutoWrapText(false)
	table.SetHeader([]string{"Name", "Configuration (protobuf text, fields with a zero value are omitted)"})
	for name, c := range configurations {
		table.Append([]string{name, proto.CompactTextString(c)})
	}
	table.Render()
	wr.Logger().Printf("%d active throttler(s) on server '%v'.\n", len(configurations), *server)
	return nil
}

func (m *UnsupportedApiMessageResponse) String() string { return proto.CompactTextString(m) }

func (m *UndeleteClusterMetadata) String() string            { return proto.CompactTextString(m) }

func (m *URLFetchResponse_Header) String() string { return proto.CompactTextString(m) }