# Azure Service Bus as a data source

> Learn how to integrate Orbital with Azure Service Bus for message streaming and publishing

Source: https://orbitalhq.com/docs/describing-data-sources/azure-service-bus

Orbital can both read messages from Azure Service Bus topics/queues to stream data as part of a query, and write messages back to Service Bus topics/queues.

## Defining a connection to Azure Service Bus

Service Bus connections are stored in your `connections.conf` config file, under the `serviceBus` element.

The connection specifies how to connect to a Service Bus namespace - but not the specifics of individual topics or queues.
(Those are specified on the Service Bus operation annotations in your Taxi model.)

```hocon
serviceBus {
   warehouseConnection {
      connectionName=warehouseConnection
      connectionParameters {
         endPoint="mynamespace.servicebus.windows.net/"
         sharedAccessKeyName="RootManageSharedAccessKey"
         sharedAccessKey="${SERVICE_BUS_ACCESS_KEY}"
      }
   }
   
   anotherConnection {
      connectionName=anotherConnection
      connectionParameters {
         endPoint="another-namespace.servicebus.windows.net/"
         sharedAccessKeyName="RootManageSharedAccessKey"
         sharedAccessKey="${ANOTHER_SERVICE_BUS_KEY}"
      }
   }
}
```

The following configuration options are supported under the `connectionParameters`:

| Config option | Purpose |
|---------------|---------|
| `endPoint` | The Service Bus namespace endpoint URL |
| `sharedAccessKeyName` | The Shared Access Signature (SAS) key name |
| `sharedAccessKey` | The Shared Access Signature (SAS) key value |

## Streaming from Azure Service Bus

You can stream messages from both Service Bus queues and topics using Orbital.

### Exposing a queue or topic subscription

Service Bus queues and topics are declared in Taxi as operations which return a `Stream` of data.

```taxi
// 1: Add the required imports
import com.orbitalhq.azure.servicebus.ServiceBusService
import com.orbitalhq.azure.servicebus.ServiceBusQueueOperation
import com.orbitalhq.azure.servicebus.ServiceBusTopicSubscriptionOperation

// Define your message model
model Person {
   name : Name inherits String
}

// 2: Annotate the service as a `ServiceBusService`
@ServiceBusService(connectionName = "warehouseConnection")
service PersonServiceBusService {
   // 3a: Stream from a queue
   @ServiceBusQueueOperation(queue = "people-queue")
   stream streamFromQueue : Stream<Person>
   
   // 3b: Stream from a topic subscription
   @ServiceBusTopicSubscriptionOperation(topic = "people-topic", subscription = "all-people")
   stream streamFromTopic : Stream<Person>
}
```

1. Add the required imports
2. Annotate the service with `@ServiceBusService`
* The `connectionName` parameter should match a connection defined in the [connections config file](#defining-a-connection-to-azure-service-bus).
3. Annotate the operations as either:
* `@ServiceBusQueueOperation` for queues, specifying the queue name
* `@ServiceBusTopicSubscriptionOperation` for topics, specifying both the topic name and subscription name
4. The return type should be a `Stream<>` of the defined message type.

### Example queries

#### Basic streaming from Service Bus
```taxi
// Stream messages from a Service Bus queue
stream { Person }
```

This will consume messages from the `people-queue` queue declared in the `PersonServiceBusService` service.

#### Enrich data from Service Bus with other sources

Data from Service Bus can be projected to enrich it with data from other sources. Data requested that is not present in the Service Bus message payload is looked up from other sources, using Orbital's standard projections.

```taxi
stream { Person } as {
  name : Name  // available in the Service Bus message
  address : Address // looked up from another data source
}[]
```

#### Filtering Service Bus streams

This example reads all messages from the Service Bus, but only emits those with a specific name pattern:

```taxi
stream { Person.filterEach( ( Person ) -> Person::Name.startsWith('J') ) }
```

#### Streaming from Service Bus to a database

Streams from Service Bus can be inserted into a database (or any other writable source) using a mutating query:

```taxi
// First, ensure that your data destination exposes a writeable data source
service DatabaseService {
   @UpsertOperation
   write operation savePerson(Person):Person
}

// Then, define a streaming query that pipes messages to the database
stream { Person }
call DatabaseService::savePerson
```

## Publishing to Azure Service Bus

To make a queue or topic writable, declare a `write operation` in a Service Bus service:

```taxi
// Add the required imports
import com.orbitalhq.azure.servicebus.ServiceBusService
import com.orbitalhq.azure.servicebus.ServiceBusQueueOperation
import com.orbitalhq.azure.servicebus.ServiceBusTopicPublicationOperation

// Define your message model
model Person {
   name : Name inherits String
}

@ServiceBusService(connectionName = "warehouseConnection")
service PersonServiceBusService {
   // Publish to a queue
   @ServiceBusQueueOperation(queue = "people-queue")
   write operation publishToQueue(Person):Person
   
   // Publish to a topic
   @ServiceBusTopicPublicationOperation(topic = "people-topic")
   write operation publishToTopic(Person):Person
}
```

### Batch publishing

Service Bus supports publishing messages in batches to improve performance. To enable batching, add the `batchSize` and `batchDuration` parameters to the operation annotation:

```taxi
@ServiceBusService(connectionName = "warehouseConnection")
service PersonServiceBusService {
   // Publish to a topic with batching
   @ServiceBusTopicPublicationOperation(
      topic = "people-topic", 
      batchSize = 5,      // Number of messages in a batch
      batchDuration = 10000  // Maximum time to wait for batch in milliseconds
   )
   write operation publishToTopic(Person):Person
}
```

When batching is configured:
1. Messages are accumulated until either:
- The batch size is reached
- The batch duration time elapses
2. The accumulated messages are then sent as a single batch to Service Bus

### Example use cases

#### Publishing a static message to Service Bus

```taxi
given { person : Person = 
  {
    name : 'John Doe'
  } 
}
call PersonServiceBusService::publishToTopic
```

#### Consuming from one Service Bus entity and publishing to another

To stream data from a Service Bus queue/topic, transform it, and republish:

```taxi
@ServiceBusService(connectionName = "warehouseConnection")
service TransformationService {
   @ServiceBusQueueOperation(queue = "raw-people")
   stream rawPeople : Stream<Person>

   @ServiceBusTopicPublicationOperation(topic = "enriched-people")
   write operation publishEnrichedPerson(EnrichedPerson):EnrichedPerson 
} 
```

The following query will consume from the `raw-people` queue, and for each message, transform it to an `EnrichedPerson`, invoking any other services required to inject the necessary data:

```taxi
stream { Person }
// Each incoming Person message is transformed to an EnrichedPerson payload
// and published to the enriched-people topic
call TransformationService::publishEnrichedPerson
```

#### Building a REST API that publishes to Service Bus

This example creates an HTTP endpoint accepting a `POST` request to publish a message to Service Bus:

```taxi
type PersonName inherits String

// The inbound request sent over HTTP
model PersonRequest {
  name : PersonName
}

// The message we'll publish to Service Bus
parameter model Person {
  name : PersonName
}

@ServiceBusService(connectionName = "warehouseConnection")
service PersonService {
   @ServiceBusTopicPublicationOperation(topic = "people-topic")
   write operation publishPerson(Person):Person
}

@HttpOperation(path = "/api/q/publishPerson", method = "POST")
query PublishPersonQuery(@RequestBody request:PersonRequest) {
  given { request }
  call PersonService::publishPerson
}
```

The above example works as follows:

* A `POST` request is sent to `/api/q/publishPerson` with a body containing a name
* The query calls `publishPerson`, which constructs a `Person` from the request
* The message is published to the Service Bus topic `people-topic`

## Advanced features

### Message properties and metadata

Service Bus provides several methods to add metadata to messages. Orbital supports accessing and setting these properties through annotations:

```taxi
import com.orbitalhq.azure.servicebus.ServiceBusMessageProperty

model Person {
   name : Name inherits String
   
   @ServiceBusMessageProperty(name = "MessageId")
   messageId : String
   
   @ServiceBusMessageProperty(name = "CorrelationId")
   correlationId : String
}
```

When consuming messages, these properties will be populated from the incoming message. When publishing messages, these properties will be set on the outgoing message.

### Error handling and dead-letter queues

Service Bus includes built-in dead-letter queue functionality. When errors occur during message processing, you can configure Orbital to handle these scenarios:

```taxi
@ServiceBusService(connectionName = "warehouseConnection")
service PersonService {
   @ServiceBusTopicSubscriptionOperation(
      topic = "people-topic", 
      subscription = "all-people",
      maxDeliveryCount = 3  // Number of processing attempts before message is dead-lettered
   )
   stream streamPeople : Stream<Person>
}
```