rabbit

Oct 25, 2023

Changes in this version

+ func ConvertDeliveryMessageToCloudevent(ctx context.Context, sourceName, namespace, queueName string, ...) (*cloudevents.Event, error)

Aug 25, 2023

Changes in this version

type RabbitMQChannelInterface

+ QueueDeclarePassive func(string, bool, bool, bool, bool, amqp091.Table) (amqp091.Queue, error)

type RabbitMQChannelMock

+ func (rm *RabbitMQChannelMock) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error)

Jan 26, 2023

Changes in this version

type BindingArgs

+ RabbitMQVhost string

type ExchangeArgs

+ RabbitMQVhost string

type QueueArgs

+ RabbitMQVhost string

type RabbitMQChannelInterface

+ IsClosed func() bool

type RabbitMQChannelMock

+ func (rm *RabbitMQChannelMock) IsClosed() bool

+ type RabbitMQConnectionHandler struct

+ Channel RabbitMQChannelInterface

+ Connection RabbitMQConnectionWrapperInterface

+ func (r *RabbitMQConnectionHandler) GetChannel() RabbitMQChannelInterface

+ func (r *RabbitMQConnectionHandler) GetConnection() RabbitMQConnectionInterface

+ func (r *RabbitMQConnectionHandler) Setup(ctx context.Context, rabbitMQURL string, ...)

type RabbitMQConnectionMock

+ NotifyCloseChannel chan *amqp.Error

+ type RabbitMQConnectionWrapperInterface interface

+ ChannelWrapper func() (RabbitMQChannelInterface, error)

+ func BadChannelDial(url string) (RabbitMQConnectionWrapperInterface, error)

+ func BadConnectionDial(url string) (RabbitMQConnectionWrapperInterface, error)

+ type RabbitMQConnectionsHandlerInterface interface

+ GetChannel func() RabbitMQChannelInterface

+ GetConnection func() RabbitMQConnectionInterface

+ Setup func(context.Context, string, ...)

+ func NewRabbitMQConnectionHandler(cycleDuration time.Duration, logger *zap.SugaredLogger) RabbitMQConnectionsHandlerInterface

Oct 21, 2022

Changes in this version

type Service

+ RabbitmqV1beta1 func() rmqv1beta1.RabbitmqV1beta1Interface

Oct 18, 2022

Changes in this version

+ const CA_SECRET_KEYNAME

+ func CloudEventToRabbitMQMessage(event *cloudevents.Event, tp, ts string) *amqp.Publishing

+ func VHostHandler(broker string, vhost string) string

type Rabbit

+ func (r *Rabbit) GetRabbitMQCASecret(ctx context.Context, clusterRef *rabbitv1beta1.RabbitmqClusterReference) (string, error)

type Service

+ GetRabbitMQCASecret func(ctx context.Context, clusterRef *rabbitv1beta1.RabbitmqClusterReference) (string, error)

Aug 23, 2022

Changes in this version

+ func ChannelConfirm(connection RabbitMQConnectionInterface, channel RabbitMQChannelInterface) error

+ func ChannelQoS(connection RabbitMQConnectionInterface, channel RabbitMQChannelInterface) error

+ func Watcher(testChannel chan bool, rabbitmqHelper RabbitMQHelper)

+ type DeleteResourceArgs struct

+ Kind interface{}

+ Name string

+ Namespace string

+ Owner metav1.Object

type Rabbit

+ func (r *Rabbit) DeleteResource(ctx context.Context, args *DeleteResourceArgs) error

+ type RabbitMQBadConnectionMock struct

+ func (rm *RabbitMQBadConnectionMock) ChannelWrapper() (RabbitMQChannelInterface, error)

+ func (rm *RabbitMQBadConnectionMock) Close() error

+ func (rm *RabbitMQBadConnectionMock) IsClosed() bool

+ func (rm *RabbitMQBadConnectionMock) NotifyClose(c chan *amqp.Error) chan *amqp.Error

+ type RabbitMQChannelInterface interface

+ Confirm func(bool) error

+ Consume func(string, string, bool, bool, bool, bool, amqp.Table) (<-chan amqp.Delivery, error)

+ NotifyClose func(chan *amqp.Error) chan *amqp.Error

+ PublishWithDeferredConfirm func(string, string, bool, bool, amqp.Publishing) (*amqp.DeferredConfirmation, error)

+ Qos func(int, int, bool) error

+ QueueInspect func(string) (amqp.Queue, error)

+ type RabbitMQChannelMock struct

+ ConsumeChannel <-chan amqp.Delivery

+ NotifyCloseChannel chan *amqp.Error

+ func (rm *RabbitMQChannelMock) Confirm(a bool) error

+ func (rm *RabbitMQChannelMock) Consume(a string, b string, c bool, d bool, e bool, f bool, t amqp.Table) (<-chan amqp.Delivery, error)

+ func (rm *RabbitMQChannelMock) NotifyClose(c chan *amqp.Error) chan *amqp.Error

+ func (rm *RabbitMQChannelMock) PublishWithDeferredConfirm(a string, b string, c bool, d bool, p amqp.Publishing) (*amqp.DeferredConfirmation, error)

+ func (rm *RabbitMQChannelMock) Qos(a int, b int, c bool) error

+ func (rm *RabbitMQChannelMock) QueueInspect(string) (amqp.Queue, error)

+ type RabbitMQConnection struct

+ func NewConnection(conn interface{}) *RabbitMQConnection

+ func (r *RabbitMQConnection) ChannelWrapper() (RabbitMQChannelInterface, error)

+ func (r *RabbitMQConnection) Close() error

+ func (r *RabbitMQConnection) IsClosed() bool

+ func (r *RabbitMQConnection) NotifyClose(c chan *amqp.Error) chan *amqp.Error

+ type RabbitMQConnectionInterface interface

+ ChannelWrapper func() (RabbitMQChannelInterface, error)

+ Close func() error

+ IsClosed func() bool

+ NotifyClose func(chan *amqp.Error) chan *amqp.Error

+ func DialWrapper(url string) (RabbitMQConnectionInterface, error)

+ func ValidConnectionDial(url string) (RabbitMQConnectionInterface, error)

+ func ValidDial(url string) (RabbitMQConnectionInterface, error)

+ type RabbitMQConnectionMock struct

+ func (rm *RabbitMQConnectionMock) ChannelWrapper() (RabbitMQChannelInterface, error)

+ func (rm *RabbitMQConnectionMock) Close() error

+ func (rm *RabbitMQConnectionMock) IsClosed() bool

+ func (rm *RabbitMQConnectionMock) NotifyClose(c chan *amqp.Error) chan *amqp.Error

type RabbitMQHelper

+ DialFunc func(string) (RabbitMQConnectionInterface, error)

+ type RabbitMQHelperInterface interface

+ CleanupRabbitMQ func(connection RabbitMQConnectionInterface, logger *zap.SugaredLogger)

+ CloseRabbitMQConnections func(RabbitMQConnectionInterface, *zap.SugaredLogger)

+ SetupRabbitMQ func(string, func(RabbitMQConnectionInterface, RabbitMQChannelInterface) error, ...) (RabbitMQConnectionInterface, RabbitMQChannelInterface, error)

+ SignalRetry func(bool)

+ WaitForRetrySignal func() bool

+ WatchRabbitMQConnections func(RabbitMQConnectionInterface, RabbitMQChannelInterface, string, ...)

type Service

+ DeleteResource func(ctx context.Context, args *DeleteResourceArgs) error

Jul 13, 2022

Changes in this version

+ const BrokerURLSecretKey

+ func MakeSecret(name, typeString, namespace, url string, owner kmeta.OwnerRefable) *corev1.Secret

+ func ReconcileSecret(ctx context.Context, secretLister corev1listers.SecretLister, ...) error

+ func SecretLabels(resourceName, typeString string) map[string]string

+ func SecretName(resourceName, typeString string) string

type QueueArgs

+ QueueType eventingv1alpha1.QueueType

type Rabbit

+ func (r *Rabbit) RabbitMQURL(ctx context.Context, clusterRef *rabbitv1beta1.RabbitmqClusterReference) (*url.URL, error)

+ func (r *Rabbit) ReconcileDLQPolicy(ctx context.Context, args *QueueArgs) (Result, error)

type Service

+ RabbitMQURL func(context.Context, *rabbitv1beta1.RabbitmqClusterReference) (*url.URL, error)

+ ReconcileDLQPolicy func(context.Context, *QueueArgs) (Result, error)

May 31, 2022

Changes in this version

+ func ConvertMessageToHTTPRequest(ctx context.Context, sourceName, namespace, queueName string, ...) error

+ func ConvertToCloudEvent(event *cloudevents.Event, msg wabbit.Delivery, ...) error

+ func NewBrokerDLXPolicy(args *QueueArgs) *rabbitv1beta1.Policy

+ type Message struct

+ ContentType string

+ Headers map[string][]byte

+ Value []byte

+ func NewMessage(value []byte, contentType string, headers map[string][]byte) *Message

+ func NewMessageFromDelivery(sourceName, namespace, queueName string, msg wabbit.Delivery) *Message

+ func (m *Message) Finish(error) error

+ func (m *Message) GetAttribute(k spec.Kind) (spec.Attribute, interface{})

+ func (m *Message) GetExtension(name string) interface{}

+ func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter) (err error)

+ func (m *Message) ReadEncoding() binding.Encoding

+ func (m *Message) ReadStructured(ctx context.Context, encoder binding.StructuredWriter) error

type QueueArgs

+ BrokerUID string

+ QueueName string

type Rabbit

+ func (r *Rabbit) ReconcileBrokerDLXPolicy(ctx context.Context, args *QueueArgs) (Result, error)

type RabbitMQHelper

+ func (r *RabbitMQHelper) WaitForRetrySignal() bool

type Service

+ ReconcileBrokerDLXPolicy func(context.Context, *QueueArgs) (Result, error)

Apr 19, 2022

Changes in this version

+ const BindingKey

+ const DLQBindingKey

+ const SourceLabelKey

+ const TriggerLabelKey

+ func Labels(b *eventingv1.Broker, t *eventingv1.Trigger, s *v1alpha1.RabbitmqSource) map[string]string

+ func NewBinding(args *BindingArgs) (*rabbitv1beta1.Binding, error)

+ func NewExchange(args *ExchangeArgs) *rabbitv1beta1.Exchange

+ func NewPolicy(args *QueueArgs) *rabbitv1beta1.Policy

+ func NewQueue(args *QueueArgs) *rabbitv1beta1.Queue

+ type BindingArgs struct

+ ClusterName string

+ Destination string

+ Filters map[string]string

+ Labels map[string]string

+ Name string

+ Namespace string

+ Owner metav1.OwnerReference

+ RabbitmqClusterReference *rabbitv1beta1.RabbitmqClusterReference

+ Source string

+ Vhost string

+ type ExchangeArgs struct

+ Broker *eventingv1.Broker

+ Name string

+ Namespace string

+ RabbitMQURL *url.URL

+ RabbitmqClusterReference *rabbitv1beta1.RabbitmqClusterReference

+ Source *v1alpha1.RabbitmqSource

+ Trigger *eventingv1.Trigger

+ type QueueArgs struct

+ DLXName *string

+ Labels map[string]string

+ Name string

+ Namespace string

+ Owner metav1.OwnerReference

+ RabbitmqClusterReference *rabbitv1beta1.RabbitmqClusterReference

+ Source *v1alpha1.RabbitmqSource

+ type RabbitMQHelper struct

+ func NewRabbitMQHelper(cycleDuration time.Duration) *RabbitMQHelper

+ func (r *RabbitMQHelper) CleanupRabbitMQ(connection *amqp.Connection, channel *amqp.Channel, retryChannel chan<- bool, ...)

+ func (r *RabbitMQHelper) CloseRabbitMQConnections(connection *amqp.Connection, channel *amqp.Channel, logger *zap.SugaredLogger)

+ func (r *RabbitMQHelper) SetupRabbitMQ(RabbitMQURL string, retryChannel chan<- bool, logger *zap.SugaredLogger) (*amqp.Connection, *amqp.Channel, error)

+ func (r *RabbitMQHelper) SignalRetry(retryChannel chan<- bool, retry bool)

+ func (r *RabbitMQHelper) WatchRabbitMQConnections(connection *amqp.Connection, channel *amqp.Channel, RabbitMQURL string, ...)

Jan 25, 2022

Changes in this version

+ type Rabbit struct

+ func New(ctx context.Context) *Rabbit

+ func (r *Rabbit) ReconcileBinding(ctx context.Context, args *triggerresources.BindingArgs) (Result, error)

+ func (r *Rabbit) ReconcileExchange(ctx context.Context, args *resources.ExchangeArgs) (Result, error)

+ func (r *Rabbit) ReconcileQueue(ctx context.Context, args *triggerresources.QueueArgs) (Result, error)

+ type Result struct

+ Name string

+ Ready bool

+ type Service interface

+ ReconcileBinding func(context.Context, *triggerresources.BindingArgs) (Result, error)

+ ReconcileExchange func(context.Context, *resources.ExchangeArgs) (Result, error)

+ ReconcileQueue func(context.Context, *triggerresources.QueueArgs) (Result, error)