You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
139 lines
3.5 KiB
139 lines
3.5 KiB
package kafka
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"net"
|
|
|
|
"github.com/segmentio/kafka-go/protocol/listgroups"
|
|
)
|
|
|
|
// ListGroupsRequest is a request to the ListGroups API.
|
|
type ListGroupsRequest struct {
|
|
// Addr is the address of the kafka broker to send the request to.
|
|
Addr net.Addr
|
|
}
|
|
|
|
// ListGroupsResponse is a response from the ListGroups API.
|
|
type ListGroupsResponse struct {
|
|
// Error is set to a non-nil value if a top-level error occurred while fetching
|
|
// groups.
|
|
Error error
|
|
|
|
// Groups contains the list of groups.
|
|
Groups []ListGroupsResponseGroup
|
|
}
|
|
|
|
// ListGroupsResponseGroup contains the response details for a single group.
|
|
type ListGroupsResponseGroup struct {
|
|
// GroupID is the ID of the group.
|
|
GroupID string
|
|
|
|
// Coordinator is the ID of the coordinator broker for the group.
|
|
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
|
|
}
|
|
|
|
// TODO: Remove everything below and use protocol-based version above everywhere.
|
|
type listGroupsRequestV1 struct {
|
|
}
|
|
|
|
func (t listGroupsRequestV1) size() int32 {
|
|
return 0
|
|
}
|
|
|
|
func (t listGroupsRequestV1) writeTo(wb *writeBuffer) {
|
|
}
|
|
|
|
type listGroupsResponseGroupV1 struct {
|
|
// GroupID holds the unique group identifier
|
|
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 holds the duration in milliseconds for which the request
|
|
// was throttled due to quota violation (Zero if the request did not violate
|
|
// any quota)
|
|
ThrottleTimeMS int32
|
|
|
|
// ErrorCode holds response error code
|
|
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
|
|
}
|
|
|