123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139 |
- package kafka
- import (
- "bufio"
- "context"
- "net"
- "github.com/segmentio/kafka-go/protocol/listgroups"
- )
- type ListGroupsRequest struct {
-
- Addr net.Addr
- }
- type ListGroupsResponse struct {
-
-
- Error error
-
- Groups []ListGroupsResponseGroup
- }
- type ListGroupsResponseGroup struct {
-
- GroupID string
-
- Coordinator int
- }
- func (c *Client) ListGroups(
- ctx context.Context,
- req *ListGroupsRequest,
- ) (*ListGroupsResponse, error) {
- protoResp, err := c.roundTrip(ctx, req.Addr, &listgroups.Request{})
- if err != nil {
- return nil, err
- }
- apiResp := protoResp.(*listgroups.Response)
- resp := &ListGroupsResponse{
- Error: makeError(apiResp.ErrorCode, ""),
- }
- for _, apiGroupInfo := range apiResp.Groups {
- resp.Groups = append(resp.Groups, ListGroupsResponseGroup{
- GroupID: apiGroupInfo.GroupID,
- Coordinator: int(apiGroupInfo.BrokerID),
- })
- }
- return resp, nil
- }
- type listGroupsRequestV1 struct {
- }
- func (t listGroupsRequestV1) size() int32 {
- return 0
- }
- func (t listGroupsRequestV1) writeTo(wb *writeBuffer) {
- }
- type listGroupsResponseGroupV1 struct {
-
- GroupID string
- ProtocolType string
- }
- func (t listGroupsResponseGroupV1) size() int32 {
- return sizeofString(t.GroupID) + sizeofString(t.ProtocolType)
- }
- func (t listGroupsResponseGroupV1) writeTo(wb *writeBuffer) {
- wb.writeString(t.GroupID)
- wb.writeString(t.ProtocolType)
- }
- func (t *listGroupsResponseGroupV1) readFrom(r *bufio.Reader, size int) (remain int, err error) {
- if remain, err = readString(r, size, &t.GroupID); err != nil {
- return
- }
- if remain, err = readString(r, remain, &t.ProtocolType); err != nil {
- return
- }
- return
- }
- type listGroupsResponseV1 struct {
-
-
-
- ThrottleTimeMS int32
-
- ErrorCode int16
- Groups []listGroupsResponseGroupV1
- }
- func (t listGroupsResponseV1) size() int32 {
- return sizeofInt32(t.ThrottleTimeMS) +
- sizeofInt16(t.ErrorCode) +
- sizeofArray(len(t.Groups), func(i int) int32 { return t.Groups[i].size() })
- }
- func (t listGroupsResponseV1) writeTo(wb *writeBuffer) {
- wb.writeInt32(t.ThrottleTimeMS)
- wb.writeInt16(t.ErrorCode)
- wb.writeArray(len(t.Groups), func(i int) { t.Groups[i].writeTo(wb) })
- }
- func (t *listGroupsResponseV1) readFrom(r *bufio.Reader, size int) (remain int, err error) {
- if remain, err = readInt32(r, size, &t.ThrottleTimeMS); err != nil {
- return
- }
- if remain, err = readInt16(r, remain, &t.ErrorCode); err != nil {
- return
- }
- fn := func(withReader *bufio.Reader, withSize int) (fnRemain int, fnErr error) {
- var item listGroupsResponseGroupV1
- if fnRemain, fnErr = (&item).readFrom(withReader, withSize); err != nil {
- return
- }
- t.Groups = append(t.Groups, item)
- return
- }
- if remain, err = readArrayWith(r, remain, fn); err != nil {
- return
- }
- return
- }
|