join_group_request.go

package sarama

type GroupProtocol struct {
	// Name contains the protocol name.
	Name string
	// Metadata contains the protocol metadata.
	Metadata []byte
}

func (p *GroupProtocol) decode(pd packetDecoder) (err error) {
	p.Name, err = pd.getString()
	if err != nil {
		return err
	}
	p.Metadata, err = pd.getBytes()
	return err
}

func (p *GroupProtocol) encode(pe packetEncoder) (err error) {
	if err := pe.putString(p.Name); err != nil {
		return err
	}
	if err := pe.putBytes(p.Metadata); err != nil {
		return err
	}
	return nil
}

type JoinGroupRequest struct {
	// Version defines the protocol version to use for encode and decode
	Version int16
	// GroupId contains the group identifier.
	GroupId string
	// SessionTimeout specifies that the coordinator should consider the consumer
	// dead if it receives no heartbeat after this timeout in milliseconds.
	SessionTimeout int32
	// RebalanceTimeout contains the maximum time in milliseconds that the
	// coordinator will wait for each member to rejoin when rebalancing the
	// group.
	RebalanceTimeout int32
	// MemberId contains the member id assigned by the group coordinator.
	MemberId string
	// GroupInstanceId contains the unique identifier of the consumer instance
	// provided by end user.
	GroupInstanceId *string
	// ProtocolType contains the unique name the for class of protocols
	// implemented by the group we want to join.
	ProtocolType string
	// GroupProtocols contains the list of protocols that the member supports.
	// deprecated; use OrderedGroupProtocols
	GroupProtocols map[string][]byte
	// OrderedGroupProtocols contains an ordered list of protocols that the member
	// supports.
	OrderedGroupProtocols []*GroupProtocol
}

func (r *JoinGroupRequest) encode(pe packetEncoder) error {
	if err := pe.putString(r.GroupId); err != nil {
		return err
	}
	pe.putInt32(r.SessionTimeout)
	if r.Version >= 1 {
		pe.putInt32(r.RebalanceTimeout)
	}
	if err := pe.putString(r.MemberId); err != nil {
		return err
	}
	if r.Version >= 5 {
		if err := pe.putNullableString(r.GroupInstanceId); err != nil {
			return err
		}
	}
	if err := pe.putString(r.ProtocolType); err != nil {
		return err
	}

	if len(r.GroupProtocols) > 0 {
		if len(r.OrderedGroupProtocols) > 0 {
			return PacketDecodingError{"cannot specify both GroupProtocols and OrderedGroupProtocols on JoinGroupRequest"}
		}

		if err := pe.putArrayLength(len(r.GroupProtocols)); err != nil {
			return err
		}
		for name, metadata := range r.GroupProtocols {
			if err := pe.putString(name); err != nil {
				return err
			}
			if err := pe.putBytes(metadata); err != nil {
				return err
			}
		}
	} else {
		if err := pe.putArrayLength(len(r.OrderedGroupProtocols)); err != nil {
			return err
		}
		for _, protocol := range r.OrderedGroupProtocols {
			if err := protocol.encode(pe); err != nil {
				return err
			}
		}
	}

	return nil
}

func (r *JoinGroupRequest) decode(pd packetDecoder, version int16) (err error) {
	r.Version = version

	if r.GroupId, err = pd.getString(); err != nil {
		return
	}

	if r.SessionTimeout, err = pd.getInt32(); err != nil {
		return
	}

	if version >= 1 {
		if r.RebalanceTimeout, err = pd.getInt32(); err != nil {
			return err
		}
	}

	if r.MemberId, err = pd.getString(); err != nil {
		return
	}

	if version >= 5 {
		if r.GroupInstanceId, err = pd.getNullableString(); err != nil {
			return
		}
	}

	if r.ProtocolType, err = pd.getString(); err != nil {
		return
	}

	n, err := pd.getArrayLength()
	if err != nil {
		return err
	}
	if n == 0 {
		return nil
	}

	r.GroupProtocols = make(map[string][]byte)
	for i := 0; i < n; i++ {
		protocol := &GroupProtocol{}
		if err := protocol.decode(pd); err != nil {
			return err
		}
		r.GroupProtocols[protocol.Name] = protocol.Metadata
		r.OrderedGroupProtocols = append(r.OrderedGroupProtocols, protocol)
	}

	return nil
}

func (r *JoinGroupRequest) key() int16 {
	return 11
}

func (r *JoinGroupRequest) version() int16 {
	return r.Version
}

func (r *JoinGroupRequest) headerVersion() int16 {
	return 1
}

func (r *JoinGroupRequest) isValidVersion() bool {
	return r.Version >= 0 && r.Version <= 5
}

func (r *JoinGroupRequest) requiredVersion() KafkaVersion {
	switch r.Version {
	case 5:
		return V2_3_0_0
	case 4:
		return V2_2_0_0
	case 3:
		return V2_0_0_0
	case 2:
		return V0_11_0_0
	case 1:
		return V0_10_1_0
	case 0:
		return V0_10_0_0
	default:
		return V2_3_0_0
	}
}

func (r *JoinGroupRequest) AddGroupProtocol(name string, metadata []byte) {
	r.OrderedGroupProtocols = append(r.OrderedGroupProtocols, &GroupProtocol{
		Name:     name,
		Metadata: metadata,
	})
}

func (r *JoinGroupRequest) AddGroupProtocolMetadata(name string, metadata *ConsumerGroupMemberMetadata) error {
	bin, err := encode(metadata, nil)
	if err != nil {
		return err
	}

	r.AddGroupProtocol(name, bin)
	return nil
}