This is an automated email from the ASF dual-hosted git repository. squakez pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel-k.git
commit 2fe006e0facc30d95241b4bb1baf4295cffa7c2b Author: Keerthan <[email protected]> AuthorDate: Wed Sep 23 21:23:52 2026 +0530 feat(binding): support ActiveMQArtemis broker direct binding in Pipe - Support binding directly to ActiveMQArtemis cluster with destination/queue property - Avoid reliance on deprecated ActiveMQArtemisAddress while keeping backward compatibility - Update unit tests and pipes.adoc documentation --- docs/modules/ROOT/pages/pipes/pipes.adoc | 36 +++++++++++-- pkg/controller/pipe/initialize_test.go | 52 +++++++++++++++++++ pkg/util/bindings/arkmq.go | 52 +++++++++++++++++-- pkg/util/bindings/arkmq_test.go | 89 +++++++++++++++++++++++++++++++- 4 files changed, 219 insertions(+), 10 deletions(-) diff --git a/docs/modules/ROOT/pages/pipes/pipes.adoc b/docs/modules/ROOT/pages/pipes/pipes.adoc index d05008d2f..ef1ca0d8f 100644 --- a/docs/modules/ROOT/pages/pipes/pipes.adoc +++ b/docs/modules/ROOT/pages/pipes/pipes.adoc @@ -179,15 +179,43 @@ KafkaTopics require the Strimzi operator and a configured KafkaTopic`. === Binding to ArkMQ queues -Another typical use case is consume/produce events directly from an ActiveMQArtemisAddress custom resource (managed by https://arkmq.org/[ArkMQ] / ActiveMQ Artemis operator): +A Pipe can consume or produce events directly using an ArkMQ / ActiveMQ Artemis broker (managed by https://arkmq.org/[ArkMQ] / ActiveMQ Artemis operator). -.beer-event-to-queue.yaml +You can bind directly to the `ActiveMQArtemis` broker resource and specify the destination queue: + +.beer-event-to-broker.yaml +[source,yaml,subs="attributes+"] +---- +apiVersion: camel.apache.org/v1 +kind: Pipe +metadata: + name: beer-event-to-broker +spec: + source: + ref: + kind: Kamelet + apiVersion: camel.apache.org/v1alpha1 + name: beer-source + properties: + period: 5000 + sink: + ref: + kind: ActiveMQArtemis + apiVersion: broker.amq.io/v1beta1 + name: my-broker + properties: + destination: beer-events +---- + +Alternatively, you can bind to an `ActiveMQArtemisAddress` custom resource: + +.beer-event-to-address.yaml [source,yaml,subs="attributes+"] ---- apiVersion: camel.apache.org/v1 kind: Pipe metadata: - name: beer-event-to-queue + name: beer-event-to-address spec: source: ref: @@ -205,7 +233,7 @@ spec: [NOTE] ==== -ActiveMQArtemisAddress requires the ActiveMQ Artemis / ArkMQ operator and an existing broker cluster. +ArkMQ bindings require the ActiveMQ Artemis / ArkMQ operator and an existing broker deployment. ==== === Binding to Knative resources diff --git a/pkg/controller/pipe/initialize_test.go b/pkg/controller/pipe/initialize_test.go index d0d9baa8b..9d4abcbd8 100644 --- a/pkg/controller/pipe/initialize_test.go +++ b/pkg/controller/pipe/initialize_test.go @@ -574,6 +574,58 @@ func TestNewPipeArkMQSourceBinding(t *testing.T) { "\"uri\":\"jms:queue:my-queue?brokerURL=tcp%3A%2F%2Fmy-broker-hdls-svc%3A61616\"},\"id\":\"binding\"}}", string(flow)) } +func TestNewPipeArkMQBrokerBinding(t *testing.T) { + pipe := &v1.Pipe{ + TypeMeta: metav1.TypeMeta{ + APIVersion: v1.SchemeGroupVersion.String(), + Kind: v1.PipeKind, + }, + ObjectMeta: metav1.ObjectMeta{ + Namespace: "ns", + Name: "my-pipe-broker", + }, + Spec: v1.PipeSpec{ + Sink: v1.Endpoint{ + Ref: &corev1.ObjectReference{ + Kind: "ActiveMQArtemis", + Name: "my-broker", + APIVersion: "broker.amq.io/v1beta1", + }, + Properties: asEndpointProperties(map[string]string{ + "destination": "my-orders", + "brokerURL": "tcp://my-broker-hdls-svc:61616", + }), + }, + Source: v1.Endpoint{ + URI: ptr.To("direct:something"), + }, + }, + } + c, err := internal.NewFakeClient(pipe) + require.NoError(t, err) + + a := NewInitializeAction() + a.InjectLogger(log.Log) + a.InjectClient(c) + assert.Equal(t, "initialize", a.Name()) + assert.True(t, a.CanHandle(pipe)) + handledPipe, err := a.Handle(context.TODO(), pipe) + require.NoError(t, err) + assert.Equal(t, v1.PipePhaseCreating, handledPipe.Status.Phase) + // Check integration which should have been created + expectedIT := v1.NewIntegration(pipe.Namespace, pipe.Name) + err = c.Get(context.Background(), ctrl.ObjectKeyFromObject(&expectedIT), &expectedIT) + require.NoError(t, err) + assert.Equal(t, pipe.Name, expectedIT.Name) + assert.Equal(t, v1.IntegrationPhaseNone, expectedIT.Status.Phase) + assert.Equal(t, "Pipe", expectedIT.Labels[kubernetes.CamelCreatorLabelKind]) + assert.Equal(t, "my-pipe-broker", expectedIT.Labels[kubernetes.CamelCreatorLabelName]) + flow, err := json.Marshal(expectedIT.Spec.Flows[0].RawMessage) + require.NoError(t, err) + assert.Equal(t, "{\"route\":{\"from\":{\"steps\":[{\"to\":\"jms:queue:my-orders?brokerURL=tcp%3A%2F%2Fmy-broker-hdls-svc%3A61616\"}],"+ + "\"uri\":\"direct:something\"},\"id\":\"binding\"}}", string(flow)) +} + func asEndpointProperties(props map[string]string) *v1.EndpointProperties { serialized, err := json.Marshal(props) if err != nil { diff --git a/pkg/util/bindings/arkmq.go b/pkg/util/bindings/arkmq.go index 94b1da06e..1e1797a8e 100644 --- a/pkg/util/bindings/arkmq.go +++ b/pkg/util/bindings/arkmq.go @@ -76,12 +76,56 @@ func (a ArkMQBindingProvider) Translate(ctx BindingContext, _ EndpointContext, e // toCamelArkMQ serializes an endpoint to a camelArkMQ struct. func (a ArkMQBindingProvider) toCamelArkMQ(ctx BindingContext, endpoint camelv1.Endpoint) (*camelArkMQ, error) { - if endpoint.Ref.Kind != arkmqv1beta1.ArkMQKindAddress { - return nil, fmt.Errorf("invalid endpoint kind %q. Can only work with %s kind", - endpoint.Ref.Kind, arkmqv1beta1.ArkMQKindAddress) + switch endpoint.Ref.Kind { + case arkmqv1beta1.ArkMQKindBroker: + return a.fromBrokerToCamel(ctx, endpoint) + case arkmqv1beta1.ArkMQKindAddress: + return a.fromAddressToCamel(ctx, endpoint) } - return a.fromAddressToCamel(ctx, endpoint) + return nil, fmt.Errorf("invalid endpoint kind %q. Can only work with %s or %s kind", + endpoint.Ref.Kind, arkmqv1beta1.ArkMQKindBroker, arkmqv1beta1.ArkMQKindAddress) +} + +// Verify and transform an ActiveMQArtemis broker resource to Camel JMS queue endpoint parameters. +func (a ArkMQBindingProvider) fromBrokerToCamel(ctx BindingContext, endpoint camelv1.Endpoint) (*camelArkMQ, error) { + props, err := endpoint.Properties.GetPropertyMap() + if err != nil { + return nil, err + } + if props == nil { + props = make(map[string]string) + } + + queueName := props["destination"] + if queueName == "" { + queueName = props["queue"] + } + if queueName == "" { + return nil, fmt.Errorf("invalid endpoint configuration: missing destination or queue property on %s %s", + endpoint.Ref.Kind, endpoint.Ref.Name) + } + delete(props, "destination") + delete(props, "queue") + + if props["brokerURL"] == "" { + namespace := endpoint.Ref.Namespace + if namespace == "" { + namespace = ctx.Namespace + } + + brokerURL, err := a.getBrokerURL(ctx, endpoint.Ref.Name, namespace) + if err != nil { + return nil, err + } + + props["brokerURL"] = brokerURL + } + + return &camelArkMQ{ + queueName: queueName, + properties: props, + }, nil } // Verify and transform an ActiveMQArtemisAddress resource to Camel JMS queue endpoint parameters. diff --git a/pkg/util/bindings/arkmq_test.go b/pkg/util/bindings/arkmq_test.go index 75289ed4c..5994f5214 100644 --- a/pkg/util/bindings/arkmq_test.go +++ b/pkg/util/bindings/arkmq_test.go @@ -223,6 +223,91 @@ func TestArkMQLookupAddressByName(t *testing.T) { assert.Equal(t, "jms:queue:events-queue?brokerURL=tcp%3A%2F%2Fmybroker-hdls-svc.test.svc%3A61616", binding.URI) } +func TestArkMQBrokerDirect(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + broker := &arkmqv1beta1.ActiveMQArtemis{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "mybroker", + }, + Status: arkmqv1beta1.ActiveMQArtemisStatus{ + PortStatus: []arkmqv1beta1.ActiveMQArtemisPortStatus{ + { + Name: "core", + Port: 61616, + }, + }, + }, + } + + client, err := internal.NewFakeClient() + require.NoError(t, err) + + bindingContext := BindingContext{ + Ctx: ctx, + Client: client, + Namespace: "test", + Profile: camelv1.TraitProfileKubernetes, + } + + // 1. With "destination" property + endpoint := camelv1.Endpoint{ + Ref: &corev1.ObjectReference{ + Kind: "ActiveMQArtemis", + Name: "mybroker", + APIVersion: "broker.amq.io/v1beta1", + }, + Properties: asEndpointProperties(map[string]string{ + "destination": "orders", + }), + } + + provider := ArkMQBindingProvider{ + Client: fake.NewSimpleClientset(broker), + } + + binding, err := provider.Translate(bindingContext, EndpointContext{ + Type: camelv1.EndpointTypeSink, + }, endpoint) + require.NoError(t, err) + assert.NotNil(t, binding) + assert.Equal(t, "jms:queue:orders?brokerURL=tcp%3A%2F%2Fmybroker-hdls-svc.test.svc%3A61616", binding.URI) + + // 2. With "queue" property fallback + endpointQueue := camelv1.Endpoint{ + Ref: &corev1.ObjectReference{ + Kind: "ActiveMQArtemis", + Name: "mybroker", + APIVersion: "broker.amq.io/v1beta1", + }, + Properties: asEndpointProperties(map[string]string{ + "queue": "invoices", + }), + } + binding, err = provider.Translate(bindingContext, EndpointContext{ + Type: camelv1.EndpointTypeSink, + }, endpointQueue) + require.NoError(t, err) + assert.NotNil(t, binding) + assert.Equal(t, "jms:queue:invoices?brokerURL=tcp%3A%2F%2Fmybroker-hdls-svc.test.svc%3A61616", binding.URI) + + // 3. Missing destination/queue property returns error + endpointMissing := camelv1.Endpoint{ + Ref: &corev1.ObjectReference{ + Kind: "ActiveMQArtemis", + Name: "mybroker", + APIVersion: "broker.amq.io/v1beta1", + }, + } + _, err = provider.Translate(bindingContext, EndpointContext{ + Type: camelv1.EndpointTypeSink, + }, endpointMissing) + require.Error(t, err) + assert.Contains(t, err.Error(), "missing destination or queue property") +} + func TestArkMQUnsupportedKind(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -237,7 +322,7 @@ func TestArkMQUnsupportedKind(t *testing.T) { Profile: camelv1.TraitProfileKubernetes, } - for _, kind := range []string{"ActiveMQArtemisQueue", "ActiveMQArtemis", "UnknownKind"} { + for _, kind := range []string{"ActiveMQArtemisQueue", "UnknownKind"} { endpoint := camelv1.Endpoint{ Ref: &corev1.ObjectReference{ Kind: kind, @@ -250,7 +335,7 @@ func TestArkMQUnsupportedKind(t *testing.T) { Type: camelv1.EndpointTypeSink, }, endpoint) require.Error(t, err) - assert.Contains(t, err.Error(), "Can only work with ActiveMQArtemisAddress kind") + assert.Contains(t, err.Error(), "Can only work with ActiveMQArtemis or ActiveMQArtemisAddress kind") } }
