mirror of
https://github.com/opencloud-eu/opencloud.git
synced 2026-06-18 12:58:50 -04:00
98 lines
3.1 KiB
Go
98 lines
3.1 KiB
Go
package kafka
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net"
|
|
"time"
|
|
|
|
"github.com/segmentio/kafka-go/protocol/describeuserscramcredentials"
|
|
)
|
|
|
|
// DescribeUserScramCredentialsRequest represents a request sent to a kafka broker to
|
|
// describe user scram credentials.
|
|
type DescribeUserScramCredentialsRequest struct {
|
|
// Address of the kafka broker to send the request to.
|
|
Addr net.Addr
|
|
|
|
// List of Scram users to describe
|
|
Users []UserScramCredentialsUser
|
|
}
|
|
|
|
type UserScramCredentialsUser struct {
|
|
Name string
|
|
}
|
|
|
|
// DescribeUserScramCredentialsResponse represents a response from a kafka broker to a describe user
|
|
// credentials request.
|
|
type DescribeUserScramCredentialsResponse struct {
|
|
// The amount of time that the broker throttled the request.
|
|
Throttle time.Duration
|
|
|
|
// Top level error that occurred while attempting to describe
|
|
// the user scram credentials.
|
|
//
|
|
// The errors contain the kafka error code. Programs may use the standard
|
|
// errors.Is function to test the error against kafka error codes.
|
|
Error error
|
|
|
|
// List of described user scram credentials.
|
|
Results []DescribeUserScramCredentialsResponseResult
|
|
}
|
|
|
|
type DescribeUserScramCredentialsResponseResult struct {
|
|
User string
|
|
CredentialInfos []DescribeUserScramCredentialsCredentialInfo
|
|
Error error
|
|
}
|
|
|
|
type DescribeUserScramCredentialsCredentialInfo struct {
|
|
Mechanism ScramMechanism
|
|
Iterations int
|
|
}
|
|
|
|
// DescribeUserScramCredentials sends a user scram credentials describe request to a kafka broker and returns
|
|
// the response.
|
|
func (c *Client) DescribeUserScramCredentials(ctx context.Context, req *DescribeUserScramCredentialsRequest) (*DescribeUserScramCredentialsResponse, error) {
|
|
users := make([]describeuserscramcredentials.RequestUser, len(req.Users))
|
|
|
|
for userIdx, user := range req.Users {
|
|
users[userIdx] = describeuserscramcredentials.RequestUser{
|
|
Name: user.Name,
|
|
}
|
|
}
|
|
|
|
m, err := c.roundTrip(ctx, req.Addr, &describeuserscramcredentials.Request{
|
|
Users: users,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("kafka.(*Client).DescribeUserScramCredentials: %w", err)
|
|
}
|
|
|
|
res := m.(*describeuserscramcredentials.Response)
|
|
responseResults := make([]DescribeUserScramCredentialsResponseResult, len(res.Results))
|
|
|
|
for responseIdx, responseResult := range res.Results {
|
|
credentialInfos := make([]DescribeUserScramCredentialsCredentialInfo, len(responseResult.CredentialInfos))
|
|
|
|
for credentialInfoIdx, credentialInfo := range responseResult.CredentialInfos {
|
|
credentialInfos[credentialInfoIdx] = DescribeUserScramCredentialsCredentialInfo{
|
|
Mechanism: ScramMechanism(credentialInfo.Mechanism),
|
|
Iterations: int(credentialInfo.Iterations),
|
|
}
|
|
}
|
|
responseResults[responseIdx] = DescribeUserScramCredentialsResponseResult{
|
|
User: responseResult.User,
|
|
CredentialInfos: credentialInfos,
|
|
Error: makeError(responseResult.ErrorCode, responseResult.ErrorMessage),
|
|
}
|
|
}
|
|
ret := &DescribeUserScramCredentialsResponse{
|
|
Throttle: makeDuration(res.ThrottleTimeMs),
|
|
Error: makeError(res.ErrorCode, res.ErrorMessage),
|
|
Results: responseResults,
|
|
}
|
|
|
|
return ret, nil
|
|
}
|