Creating a delivery and sending data
This example demonstrates the complete workflow for creating a delivery and sending data through the Local Actor API v2. The workflow follows the pattern: CREATE → POLL → CONNECT → USE, where you first create a delivery via the REST API, poll for its status until it's ready, then connect to the provided AMQP endpoint to publish messages.
Prerequisites
General Requirements
- Valid client certificate and private key in PEM format
- CA certificate in PEM format
- Access to a Local Actor API v2 instance
Language-Specific Requirements
- Python 3.x
- python-qpid-proton library
- requests library
- .NET 6.0 or later
- AMQPNetLite NuGet package
- Go 1.18 or later
- github.com/Azure/go-amqp package
- Java 11 or later
- Maven 3.6 or later
- Apache Qpid Proton-J library
- OkHttp library
- Jackson library
- Bouncy Castle library
- Node.js 16 or later
- rhea library
Environment Variables
| Variable | Description | Example |
|---|---|---|
ACTOR_API_HOST |
Hostname of the Actor API instance | api.example.com |
ACTOR_API_PORT |
Port of the Actor API instance | 443 |
ACTOR_API_DELIVERY_SELECTOR |
Selector for the delivery to create | messageType = 'TEST' |
ACTOR_COMMON_NAME |
Common name from the actor client certificate | actor.example.com |
ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM |
Path to client certificate and private key in PEM format | /path/to/client-cert-and-key.pem |
CA_CERTIFICATE_PEM |
Path to CA certificate in PEM format | /path/to/ca-cert.pem |
MESSAGE_APPLICATION_PROPERTIES_JSON |
Message properties in JSON format | {"messageType": "TEST", "publisherId": "XX99999", "publicationId": "XX99999:TEST", "originatingCountry": "XX", "protocolVersion": "TEST:0.0.0", "quadTree": ",1004,"} |
Configuration
ACTOR_API_HOST=os.environ.get("ACTOR_API_HOST", "hostname_of_the_actor_api")
ACTOR_API_PORT=os.environ.get("ACTOR_API_PORT", "port_of_the_actor_api")
ACTOR_API_DELIVERY_SELECTOR=os.environ.get("ACTOR_API_DELIVERY_SELECTOR", "selector_of_the_delivery")
ACTOR_COMMON_NAME=os.environ.get("ACTOR_COMMON_NAME", "cn_of_the_actor_client_certificate")
ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM=os.environ.get("ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM", "pem_with_x509_certificate_chain_and_private_key")
CA_CERTIFICATE_PEM=os.environ.get("CA_CERTIFICATE_PEM", "pem_with_x509_certificate")
MESSAGE_APPLICATION_PROPERTIES_JSON=os.environ.get("MESSAGE_APPLICATION_PROPERTIES_JSON", "message_application_properties_json")
private static readonly string ACTOR_API_HOST = Environment.GetEnvironmentVariable("ACTOR_API_HOST") ?? "hostname_of_the_actor_api";
private static readonly string ACTOR_API_PORT = Environment.GetEnvironmentVariable("ACTOR_API_PORT") ?? "port_of_the_actor_api";
private static readonly string ACTOR_API_DELIVERY_SELECTOR = Environment.GetEnvironmentVariable("ACTOR_API_DELIVERY_SELECTOR") ?? "selector_of_the_delivery";
private static readonly string ACTOR_COMMON_NAME = Environment.GetEnvironmentVariable("ACTOR_COMMON_NAME") ?? "cn_of_the_actor_client_certificate";
private static readonly string ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM = Environment.GetEnvironmentVariable("ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM") ?? "pem_with_x509_certificate_chain_and_private_key";
private static readonly string CA_CERTIFICATE_PEM = Environment.GetEnvironmentVariable("CA_CERTIFICATE_PEM") ?? "pem_with_x509_certificate";
private static readonly string MESSAGE_APPLICATION_PROPERTIES_JSON = Environment.GetEnvironmentVariable("MESSAGE_APPLICATION_PROPERTIES_JSON") ?? "message_application_properties_json";
var (
ACTOR_API_HOST = getEnv("ACTOR_API_HOST", "hostname_of_the_actor_api")
ACTOR_API_PORT = getEnv("ACTOR_API_PORT", "port_of_the_actor_api")
ACTOR_API_DELIVERY_SELECTOR = getEnv("ACTOR_API_DELIVERY_SELECTOR", "selector_of_the_delivery")
ACTOR_COMMON_NAME = getEnv("ACTOR_COMMON_NAME", "cn_of_the_actor_client_certificate")
ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM = getEnv("ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM", "pem_with_x509_certificate_chain_and_private_key")
CA_CERTIFICATE_PEM = getEnv("CA_CERTIFICATE_PEM", "pem_with_x509_certificate")
MESSAGE_APPLICATION_PROPERTIES_JSON = getEnv("MESSAGE_APPLICATION_PROPERTIES_JSON", "message_application_properties_json")
)
func getEnv(key, defaultValue string) string {
if value := os.Getenv(key); value != "" {
return value
}
return defaultValue
}
private static final String ACTOR_API_HOST = getEnv("ACTOR_API_HOST", "hostname_of_the_actor_api");
private static final String ACTOR_API_PORT = getEnv("ACTOR_API_PORT", "port_of_the_actor_api");
private static final String ACTOR_API_DELIVERY_SELECTOR = getEnv("ACTOR_API_DELIVERY_SELECTOR", "selector_of_the_delivery");
private static final String ACTOR_COMMON_NAME = getEnv("ACTOR_COMMON_NAME", "cn_of_the_actor_client_certificate");
private static final String ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM = getEnv("ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM", "pem_with_x509_certificate_chain_and_private_key");
private static final String CA_CERTIFICATE_PEM = getEnv("CA_CERTIFICATE_PEM", "pem_with_x509_certificate");
private static final String MESSAGE_APPLICATION_PROPERTIES_JSON = getEnv("MESSAGE_APPLICATION_PROPERTIES_JSON", "message_application_properties_json");
private static final ObjectMapper objectMapper = new ObjectMapper();
private static OkHttpClient httpClient;
private static String getEnv(String key, String defaultValue) {
String value = System.getenv(key);
return value != null ? value : defaultValue;
}
const ACTOR_API_HOST = process.env.ACTOR_API_HOST || 'hostname_of_the_actor_api';
const ACTOR_API_PORT = process.env.ACTOR_API_PORT || 'port_of_the_actor_api';
const ACTOR_API_DELIVERY_SELECTOR = process.env.ACTOR_API_DELIVERY_SELECTOR || 'selector_of_the_delivery';
const ACTOR_COMMON_NAME = process.env.ACTOR_COMMON_NAME || 'cn_of_the_actor_client_certificate';
const ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM = process.env.ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM || 'pem_with_x509_certificate_chain_and_private_key';
const CA_CERTIFICATE_PEM = process.env.CA_CERTIFICATE_PEM || 'pem_with_x509_certificate';
const MESSAGE_APPLICATION_PROPERTIES_JSON = process.env.MESSAGE_APPLICATION_PROPERTIES_JSON || 'message_application_properties_json';
// Message counter
let messageCount = 0;
Explanation
Configuration by environment variables
API Functions
def api_url(endpoint):
return "https://%s:%s/%s/%s" % (ACTOR_API_HOST, ACTOR_API_PORT, ACTOR_COMMON_NAME, endpoint)
def api_get(endpoint):
return requests.get(api_url(endpoint), verify=CA_CERTIFICATE_PEM, cert=ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM)
def api_post(endpoint, json_data):
return requests.post(api_url(endpoint), None, json_data, verify=CA_CERTIFICATE_PEM, cert=ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM)
def api_delete(endpoint):
return requests.delete(api_url(endpoint), verify=CA_CERTIFICATE_PEM, cert=ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM)
def api_get_delivery(id):
return api_get("deliveries/%s" % id)
def api_delete_delivery(id):
return api_delete("deliveries/%s" % id)
def api_create_delivery():
json_data = {
"selector": ACTOR_API_DELIVERY_SELECTOR
}
return api_post("deliveries", json_data)
private static HttpClient CreateHttpClient()
{
var handler = new HttpClientHandler();
try
{
// Load client certificate using the same method as AMQP
var certAndKeyPem = File.ReadAllText(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM);
var clientCert = X509Certificate2.CreateFromPem(certAndKeyPem, certAndKeyPem);
handler.ClientCertificates.Add(clientCert);
// Load CA certificate for server validation
var caCertPem = File.ReadAllText(CA_CERTIFICATE_PEM);
var caCert = X509Certificate2.CreateFromPem(caCertPem);
handler.ServerCertificateCustomValidationCallback = (sender, cert, chain, errors) => {
return chain.ChainElements[chain.ChainElements.Count - 1].Certificate.Equals(caCert);
};
}
catch (Exception ex)
{
LogError($"Error loading certificates for HTTP client: {ex.Message}");
throw;
}
return new HttpClient(handler);
}
private static string ApiUrl(string endpoint)
{
return $"https://{ACTOR_API_HOST}:{ACTOR_API_PORT}/{ACTOR_COMMON_NAME}/{endpoint}";
}
private static async Task<HttpResponseMessage> ApiGetAsync(string endpoint, HttpClient client)
{
return await client.GetAsync(ApiUrl(endpoint));
}
private static async Task<HttpResponseMessage> ApiPostAsync(string endpoint, object jsonData, HttpClient client)
{
var json = JsonSerializer.Serialize(jsonData);
var content = new StringContent(json, Encoding.UTF8, "application/json");
return await client.PostAsync(ApiUrl(endpoint), content);
}
private static async Task<HttpResponseMessage> ApiDeleteAsync(string endpoint, HttpClient client)
{
return await client.DeleteAsync(ApiUrl(endpoint));
}
private static async Task<HttpResponseMessage> ApiGetDeliveryAsync(string id, HttpClient client)
{
return await ApiGetAsync($"deliveries/{id}", client);
}
private static async Task<HttpResponseMessage> ApiDeleteDeliveryAsync(string id, HttpClient client)
{
return await ApiDeleteAsync($"deliveries/{id}", client);
}
private static async Task<HttpResponseMessage> ApiCreateDeliveryAsync(HttpClient client)
{
var jsonData = new { selector = ACTOR_API_DELIVERY_SELECTOR };
return await ApiPostAsync("deliveries", jsonData, client);
}
func apiURL(endpoint string) string {
return fmt.Sprintf("https://%s:%s/%s/%s", ACTOR_API_HOST, ACTOR_API_PORT, ACTOR_COMMON_NAME, endpoint)
}
func createHTTPClient() (*http.Client, error) {
// Load client certificate and key
cert, err := tls.LoadX509KeyPair(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM, ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM)
if err != nil {
return nil, fmt.Errorf("failed to load client certificate: %v", err)
}
// Load CA certificate
caCert, err := os.ReadFile(CA_CERTIFICATE_PEM)
if err != nil {
return nil, fmt.Errorf("failed to read CA certificate: %v", err)
}
// Create certificate pool with CA
caCertPool := x509.NewCertPool()
caCertPool.AppendCertsFromPEM(caCert)
tlsConfig := &tls.Config{
Certificates: []tls.Certificate{cert},
RootCAs: caCertPool,
MinVersion: tls.VersionTLS13,
}
client := &http.Client{
Transport: &http.Transport{
TLSClientConfig: tlsConfig,
},
}
return client, nil
}
func apiGet(endpoint string) (*http.Response, error) {
client, err := createHTTPClient()
if err != nil {
return nil, err
}
return client.Get(apiURL(endpoint))
}
func apiPost(endpoint string, jsonData map[string]interface{}) (*http.Response, error) {
client, err := createHTTPClient()
if err != nil {
return nil, err
}
jsonBytes, err := json.Marshal(jsonData)
if err != nil {
return nil, err
}
return client.Post(apiURL(endpoint), "application/json", strings.NewReader(string(jsonBytes)))
}
func apiDelete(endpoint string) (*http.Response, error) {
client, err := createHTTPClient()
if err != nil {
return nil, err
}
req, err := http.NewRequest("DELETE", apiURL(endpoint), nil)
if err != nil {
return nil, err
}
return client.Do(req)
}
func apiGetDelivery(id string) (*http.Response, error) {
return apiGet(fmt.Sprintf("deliveries/%s", id))
}
func apiDeleteDelivery(id string) (*http.Response, error) {
return apiDelete(fmt.Sprintf("deliveries/%s", id))
}
func apiCreateDelivery() (*http.Response, error) {
jsonData := map[string]interface{}{
"selector": ACTOR_API_DELIVERY_SELECTOR,
}
return apiPost("deliveries", jsonData)
}
private static String apiUrl(String endpoint) {
return String.format("https://%s:%s/%s/%s", ACTOR_API_HOST, ACTOR_API_PORT, ACTOR_COMMON_NAME, endpoint);
}
private static Response apiGet(String endpoint) throws IOException {
Request request = new Request.Builder()
.url(apiUrl(endpoint))
.build();
return httpClient.newCall(request).execute();
}
private static Response apiPost(String endpoint, String jsonData) throws IOException {
RequestBody body = RequestBody.create(jsonData, MediaType.parse("application/json"));
Request request = new Request.Builder()
.url(apiUrl(endpoint))
.post(body)
.build();
return httpClient.newCall(request).execute();
}
private static Response apiDelete(String endpoint) throws IOException {
Request request = new Request.Builder()
.url(apiUrl(endpoint))
.delete()
.build();
return httpClient.newCall(request).execute();
}
private static Response apiGetDelivery(String id) throws IOException {
return apiGet("deliveries/" + id);
}
private static Response apiDeleteDelivery(String id) throws IOException {
return apiDelete("deliveries/" + id);
}
private static Response apiCreateDelivery() throws IOException {
Map<String, Object> jsonData = new HashMap<>();
jsonData.put("selector", ACTOR_API_DELIVERY_SELECTOR);
String json = objectMapper.writeValueAsString(jsonData);
return apiPost("deliveries", json);
}
function apiUrl(endpoint) {
return `https://${ACTOR_API_HOST}:${ACTOR_API_PORT}/${ACTOR_COMMON_NAME}/${endpoint}`;
}
function createHttpsAgent() {
return new https.Agent({
cert: fs.readFileSync(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM),
key: fs.readFileSync(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM),
ca: [fs.readFileSync(CA_CERTIFICATE_PEM)],
rejectUnauthorized: true
});
}
function apiRequest(method, endpoint, data = null) {
return new Promise((resolve, reject) => {
const url = new URL(apiUrl(endpoint));
const agent = createHttpsAgent();
const options = {
hostname: url.hostname,
port: url.port,
path: url.pathname,
method: method,
agent: agent,
headers: {
'Content-Type': 'application/json'
}
};
const req = https.request(options, (res) => {
let responseData = '';
res.on('data', (chunk) => {
responseData += chunk;
});
res.on('end', () => {
const response = {
statusCode: res.statusCode,
ok: res.statusCode >= 200 && res.statusCode < 300,
data: responseData,
json: () => JSON.parse(responseData)
};
resolve(response);
});
});
req.on('error', (error) => {
reject(error);
});
if (data) {
req.write(JSON.stringify(data));
}
req.end();
});
}
function apiGet(endpoint) {
return apiRequest('GET', endpoint);
}
function apiPost(endpoint, jsonData) {
return apiRequest('POST', endpoint, jsonData);
}
function apiDelete(endpoint) {
return apiRequest('DELETE', endpoint);
}
function apiGetDelivery(id) {
return apiGet(`deliveries/${id}`);
}
function apiDeleteDelivery(id) {
return apiDelete(`deliveries/${id}`);
}
function apiCreateDelivery() {
const jsonData = {
selector: ACTOR_API_DELIVERY_SELECTOR
};
return apiPost('deliveries', jsonData);
}
Explanation
Implementation of the required delivery endpoints (see REST API Reference)
AMQP 1.0 Client
def amqp_create_ssl_config():
ssl_config = SSLDomain(SSLDomain.MODE_CLIENT)
ssl_config.set_peer_authentication(SSLDomain.ANONYMOUS_PEER)
ssl_config.set_credentials(cert_file=ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM, key_file=ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM, password=None)
ssl_config.set_trusted_ca_db(CA_CERTIFICATE_PEM)
return ssl_config
class Sender(MessagingHandler):
def __init__(self, endpoint):
super(Sender, self).__init__()
self.__endpoint = endpoint
self.message_count = 0
def on_start(self, event):
logging.debug("Container reactor started")
container = event.container
endpoint = self.__endpoint
# Step 1: connect
ssl_config = amqp_create_ssl_config()
amqp_url = "amqps://%s:%s" % (endpoint["host"], endpoint["port"])
connection = container.connect(amqp_url, ssl_domain = ssl_config, reconnect = False, heartbeat = 5)
# Step 2: create a sending link using the target address of the endpoint
self.sender_link = container.create_sender(connection, endpoint["target"])
def send_message(self):
# Increment message counter
self.message_count += 1
# Create dynamic message content with counter and timestamp
body_text = f"Hello World! Message #{self.message_count} at {datetime.now().strftime('%H:%M:%S')}"
body_binary = body_text.encode('utf-8')
message = Message()
# Make sure to use the Data type for the body to ensure it is sent as binary
message.inferred = True
message.body = body_binary
message.properties = json.loads(MESSAGE_APPLICATION_PROPERTIES_JSON)
# Format properties in sorted order for consistent logging
properties_dict = json.loads(MESSAGE_APPLICATION_PROPERTIES_JSON)
sorted_properties = json.dumps(properties_dict, sort_keys=True)
logging.info("Sending message: body='%s', properties=%s", body_text, sorted_properties)
self.sender_link.send(message)
def on_link_opened(self, event):
# Send first message when link is opened (like Java onLinkRemoteOpen)
if hasattr(event.link, 'credit') and event.link.credit > 0:
self.send_message()
def on_settled(self, event):
# Schedule next message after 1 second
event.container.schedule(1.0, self)
def on_timer_task(self, event):
# Send next message
self.send_message()
def amqp_connect_and_publish(endpoint):
sender = Sender(endpoint)
container = Container(sender)
thread = threading.Thread(name = "AMQPClient", target = container.run, daemon = True)
thread.start()
while thread.is_alive():
time.sleep(1)
private static ConnectionFactory CreateConnectionFactory()
{
var factory = new ConnectionFactory();
try
{
// Configure SSL/TLS with client certificate for SASL EXTERNAL
// Read the combined PEM file content
var certAndKeyPem = File.ReadAllText(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM);
var clientCert = X509Certificate2.CreateFromPem(certAndKeyPem, certAndKeyPem);
factory.SSL.ClientCertificates.Add(clientCert);
// Enable SSL/TLS
factory.SSL.Protocols = System.Security.Authentication.SslProtocols.Tls13;
// Load CA certificate for validation
var caCertPem = File.ReadAllText(CA_CERTIFICATE_PEM);
var caCert = X509Certificate2.CreateFromPem(caCertPem);
factory.SSL.RemoteCertificateValidationCallback = (sender, cert, chain, errors) => {
// Validate against CA certificate
return ValidateCertificate(cert as X509Certificate2, caCert);
};
// Extract common name for SASL EXTERNAL
var commonName = ExtractCommonName(clientCert.Subject);
LogDebug($"Certificate Subject: {clientCert.Subject}");
LogDebug($"Extracted Common Name: {commonName}");
// Configure custom SASL EXTERNAL profile that sends only the CN value
factory.SASL.Profile = new CustomSaslExternalProfile(commonName);
}
catch (Exception ex)
{
LogError($"Error loading certificates: {ex.Message}");
throw;
}
return factory;
}
private static bool ValidateCertificate(X509Certificate2? serverCert, X509Certificate2 caCert)
{
if (serverCert == null || caCert == null)
return false;
var chain = new X509Chain();
chain.ChainPolicy.ExtraStore.Add(caCert);
chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck;
chain.ChainPolicy.VerificationFlags = X509VerificationFlags.AllowUnknownCertificateAuthority;
return chain.Build(serverCert);
}
private static string ExtractCommonName(string subjectDn)
{
// Extract CN value from Distinguished Name (e.g., "CN=XX99999, O=Company" -> "XX99999")
var parts = subjectDn.Split(',');
foreach (var part in parts)
{
var trimmed = part.Trim();
if (trimmed.StartsWith("CN="))
{
return trimmed.Substring(3);
}
}
return subjectDn; // Fallback to full DN if CN not found
}
// Custom SASL EXTERNAL profile that sends only the CN value
public class CustomSaslExternalProfile : SaslProfile
{
private readonly string identity;
public CustomSaslExternalProfile(string identity) : base(new Symbol("EXTERNAL"))
{
this.identity = identity;
}
protected override DescribedList GetStartCommand(string hostname)
{
return new SaslInit()
{
Mechanism = "EXTERNAL",
InitialResponse = Encoding.UTF8.GetBytes(identity)
};
}
protected override DescribedList OnCommand(DescribedList command)
{
// For EXTERNAL, we should only need the initial response
return null!;
}
protected override ITransport UpgradeTransport(ITransport transport)
{
// No transport upgrade needed for EXTERNAL
return transport;
}
}
private static async Task AmqpConnectAndPublishAsync(DeliveryEndpoint endpoint)
{
var factory = CreateConnectionFactory();
// Extract CN from certificate for user identity
var certAndKeyPem = File.ReadAllText(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM);
var clientCert = X509Certificate2.CreateFromPem(certAndKeyPem, certAndKeyPem);
var commonName = ExtractCommonName(clientCert.Subject);
// Create connection with proper user identity
var address = new Address($"amqps://{endpoint.Host}:{endpoint.Port}");
var connection = await factory.CreateAsync(address);
var session = new Session(connection);
var sender = new SenderLink(session, "sender-link", endpoint.Target);
var messageProperties = JsonSerializer.Deserialize<Dictionary<string, JsonElement>>(MESSAGE_APPLICATION_PROPERTIES_JSON);
LogInfo("Starting to send messages continuously. Press Ctrl+C to stop.");
var cts = new CancellationTokenSource();
Console.CancelKeyPress += (sender_event, e) => {
e.Cancel = true;
cts.Cancel();
};
int messageCount = 0;
try
{
while (!cts.Token.IsCancellationRequested)
{
// Increment message counter
messageCount++;
// Create dynamic message content with counter and timestamp
var bodyText = $"Hello World! Message #{messageCount} at {DateTime.Now:HH:mm:ss}";
var bodyBinary = Encoding.UTF8.GetBytes(bodyText);
var message = new Message()
{
BodySection = new Data() { Binary = bodyBinary },
ApplicationProperties = new ApplicationProperties()
};
foreach (var prop in messageProperties)
{
// Convert JsonElement to proper .NET types for AMQP
object value = prop.Value.ValueKind switch
{
JsonValueKind.String => prop.Value.GetString(),
JsonValueKind.Number => prop.Value.GetInt32(),
JsonValueKind.True => true,
JsonValueKind.False => false,
JsonValueKind.Null => null,
_ => prop.Value.ToString()
};
message.ApplicationProperties[prop.Key] = value;
}
// Format application properties for logging
var appPropsString = "";
if (message.ApplicationProperties != null && message.ApplicationProperties.Map != null)
{
var props = message.ApplicationProperties.Map.Select(kvp => $"{kvp.Key}={kvp.Value}");
appPropsString = " [" + string.Join(", ", props) + "]";
}
LogInfo($"Sending message #{messageCount}: {message.Body}{appPropsString}");
await sender.SendAsync(message);
// Wait 1 second before sending the next message
await Task.Delay(1000, cts.Token);
}
}
catch (OperationCanceledException)
{
LogInfo($"Stopping message sending... Sent {messageCount} messages total.");
}
await sender.CloseAsync();
await session.CloseAsync();
await connection.CloseAsync();
}
func amqpCreateTLSConfig() (*tls.Config, error) {
// Load client certificate and key
cert, err := tls.LoadX509KeyPair(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM, ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM)
if err != nil {
return nil, fmt.Errorf("failed to load client certificate: %v", err)
}
// Load CA certificate
caCert, err := os.ReadFile(CA_CERTIFICATE_PEM)
if err != nil {
return nil, fmt.Errorf("failed to read CA certificate: %v", err)
}
// Create certificate pool with CA
caCertPool := x509.NewCertPool()
caCertPool.AppendCertsFromPEM(caCert)
tlsConfig := &tls.Config{
Certificates: []tls.Certificate{cert},
RootCAs: caCertPool,
InsecureSkipVerify: false,
MinVersion: tls.VersionTLS13,
}
return tlsConfig, nil
}
type Sender struct {
client *amqp.Conn
session *amqp.Session
sender *amqp.Sender
endpoint map[string]interface{}
sending bool
messageCount int
}
func (s *Sender) connect() error {
// Step 1: Create TLS configuration
tlsConfig, err := amqpCreateTLSConfig()
if err != nil {
return err
}
// Create AMQP URL
host := s.endpoint["host"].(string)
port := s.endpoint["port"]
amqpURL := fmt.Sprintf("amqps://%s:%v", host, port)
log.Printf("Connecting to %s", amqpURL)
// Step 2: Connect with TLS
opts := &amqp.ConnOptions{
TLSConfig: tlsConfig,
SASLType: amqp.SASLTypeExternal(""),
IdleTimeout: 0,
MaxFrameSize: 65536,
ContainerID: "",
HostName: host,
Properties: nil,
}
ctx := context.Background()
client, err := amqp.Dial(ctx, amqpURL, opts)
if err != nil {
return fmt.Errorf("failed to connect: %v", err)
}
s.client = client
// Create session
session, err := client.NewSession(ctx, nil)
if err != nil {
return fmt.Errorf("failed to create session: %v", err)
}
s.session = session
// Step 3: Create sender link using the target address
target := s.endpoint["target"].(string)
sender, err := session.NewSender(ctx, target, nil)
if err != nil {
return fmt.Errorf("failed to create sender: %v", err)
}
s.sender = sender
log.Println("Container reactor started")
s.sending = true
return nil
}
func (s *Sender) sendMessage() error {
// Parse message properties from JSON
var properties map[string]interface{}
err := json.Unmarshal([]byte(MESSAGE_APPLICATION_PROPERTIES_JSON), &properties)
if err != nil {
return fmt.Errorf("failed to parse message properties: %v", err)
}
// Increment message counter
s.messageCount++
// Create dynamic message content with counter and timestamp
bodyText := fmt.Sprintf("Hello World! Message #%d at %s", s.messageCount, time.Now().Format("15:04:05"))
bodyBinary := []byte(bodyText)
// Create message
msg := &amqp.Message{
Data: [][]byte{bodyBinary},
ApplicationProperties: properties,
}
ctx := context.Background()
// Format message for logging with sorted properties
propsMap := make(map[string]interface{})
for k, v := range properties {
propsMap[k] = v
}
propsJSON, _ := json.Marshal(propsMap)
log.Printf("Sending message: body='%s', properties=%s", bodyText, string(propsJSON))
// Send message
err = s.sender.Send(ctx, msg, nil)
if err != nil {
return fmt.Errorf("failed to send message: %v", err)
}
log.Printf("Message settled")
return nil
}
func (s *Sender) run() {
for s.sending {
err := s.sendMessage()
if err != nil {
log.Printf("Error sending message: %v", err)
}
time.Sleep(1 * time.Second)
}
}
func (s *Sender) close() {
s.sending = false
if s.sender != nil {
s.sender.Close(context.Background())
}
if s.session != nil {
s.session.Close(context.Background())
}
if s.client != nil {
s.client.Close()
}
}
func amqpConnectAndPublish(endpoint map[string]interface{}) error {
sender := &Sender{
endpoint: endpoint,
}
err := sender.connect()
if err != nil {
return err
}
defer sender.close()
// Run sender
sender.run()
return nil
}
private static class SenderHandler extends BaseHandler {
private final Map<String, Object> endpoint;
private final AtomicInteger messageCount = new AtomicInteger(0);
private Sender sender;
private final SimpleDateFormat timeFormat = new SimpleDateFormat("HH:mm:ss");
private SSLContext sslContext;
public SenderHandler(Map<String, Object> endpoint) {
this.endpoint = endpoint;
}
public void setSslContext(SSLContext sslContext) {
this.sslContext = sslContext;
}
@Override
public void onConnectionInit(Event event) {
logger.fine("Connection initialized");
Connection connection = event.getConnection();
connection.setHostname((String) endpoint.get("host"));
connection.setContainer("java-delivery-example");
connection.open();
}
@Override
public void onConnectionBound(Event event) {
logger.fine("Connection bound, configuring transport");
Transport transport = event.getTransport();
if (sslContext != null) {
// Configure SASL EXTERNAL
Sasl sasl = transport.sasl();
sasl.setMechanisms("EXTERNAL");
// Configure SSL
SslDomain sslDomain = Proton.sslDomain();
sslDomain.init(SslDomain.Mode.CLIENT);
sslDomain.setPeerAuthentication(SslDomain.VerifyMode.VERIFY_PEER);
sslDomain.setSslContext(sslContext);
transport.ssl(sslDomain);
}
}
@Override
public void onConnectionRemoteOpen(Event event) {
logger.fine("Connection opened");
Connection connection = event.getConnection();
Session session = connection.session();
session.open();
}
@Override
public void onSessionRemoteOpen(Event event) {
logger.fine("Session opened");
Session session = event.getSession();
Target target = new Target();
String targetAddress = (String) endpoint.get("target");
target.setAddress(targetAddress);
sender = session.sender(targetAddress);
sender.setTarget(target);
Source source = new Source();
sender.setSource(source);
sender.open();
}
@Override
public void onLinkRemoteClose(Event event) {
Link link = event.getLink();
logger.severe("Link remote close - State: " + link.getRemoteState());
if (link.getRemoteCondition() != null) {
logger.severe("Condition: " + link.getRemoteCondition().getCondition());
logger.severe("Description: " + link.getRemoteCondition().getDescription());
}
}
@Override
public void onLinkRemoteOpen(Event event) {
logger.fine("Sender link opened, ready to send messages");
if (event.getLink() instanceof Sender && event.getSender().getCredit() > 0) {
sendMessage();
}
}
@Override
public void onDelivery(Event event) {
Delivery delivery = event.getDelivery();
if (delivery.getRemoteState() != null) {
delivery.settle();
// Schedule next message after 1 second
event.getReactor().schedule(1000, this);
}
}
@Override
public void onTimerTask(Event event) {
if (sender != null && sender.getCredit() > 0) {
sendMessage();
}
}
private void sendMessage() {
try {
// Increment message counter
int count = messageCount.incrementAndGet();
// Create dynamic message content with counter and timestamp
String bodyText = String.format("Hello World! Message #%d at %s", count, timeFormat.format(new Date()));
// Create message
Message message = Message.Factory.create();
message.setBody(new Data(new Binary(bodyText.getBytes(StandardCharsets.UTF_8))));
// Parse and set application properties
Map<String, Object> properties = objectMapper.readValue(MESSAGE_APPLICATION_PROPERTIES_JSON,
new TypeReference<Map<String, Object>>() {
});
message.setApplicationProperties(new ApplicationProperties(properties));
// Format properties for logging
String sortedProperties = objectMapper.writeValueAsString(new TreeMap<>(properties));
logger.info(String.format("Sending message: body='%s', properties=%s", bodyText, sortedProperties));
// Send message
byte[] encodedMessage = new byte[1024];
int encodedSize = message.encode(encodedMessage, 0, encodedMessage.length);
Delivery delivery = sender.delivery(new byte[0]);
sender.send(encodedMessage, 0, encodedSize);
sender.advance();
} catch (Exception e) {
logger.log(Level.WARNING, "Error sending message", e);
e.printStackTrace();
}
}
@Override
public void onTransportError(Event event) {
logger.log(Level.SEVERE, "Transport error: " + event.getTransport().getCondition());
}
}
private static SSLContext createSSLContext() throws Exception {
// Add BouncyCastle provider
Security.addProvider(new BouncyCastleProvider());
// Create SSL context
SSLContext sslContext = SSLContext.getInstance("TLSv1.3");
// Load client certificate and key
KeyStore keyStore = KeyStore.getInstance("PKCS12");
keyStore.load(null);
// Parse certificates and key using BouncyCastle
List<java.security.cert.Certificate> certChain = new ArrayList<>();
PrivateKey privateKey = null;
try (PEMParser pemParser = new PEMParser(new FileReader(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM))) {
Object object;
JcaPEMKeyConverter keyConverter = new JcaPEMKeyConverter().setProvider("BC");
JcaX509CertificateConverter certConverter = new JcaX509CertificateConverter().setProvider("BC");
while ((object = pemParser.readObject()) != null) {
if (object instanceof X509CertificateHolder) {
certChain.add(certConverter.getCertificate((X509CertificateHolder) object));
} else if (object instanceof PEMKeyPair) {
privateKey = keyConverter.getPrivateKey(((PEMKeyPair) object).getPrivateKeyInfo());
}
}
}
// Add certificate chain and key to keystore
if (privateKey != null && !certChain.isEmpty()) {
keyStore.setKeyEntry("client", privateKey, new char[0], certChain.toArray(new java.security.cert.Certificate[0]));
} else {
throw new IllegalStateException("Failed to load client certificate and key");
}
// Initialize key manager
KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
kmf.init(keyStore, new char[0]);
// Load CA certificate
KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
trustStore.load(null);
CertificateFactory cf = CertificateFactory.getInstance("X.509");
try (FileInputStream fis = new FileInputStream(CA_CERTIFICATE_PEM)) {
java.security.cert.Certificate caCert = cf.generateCertificate(fis);
trustStore.setCertificateEntry("ca", caCert);
}
// Initialize trust manager
TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(trustStore);
// Initialize SSL context
sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), new SecureRandom());
return sslContext;
}
private static void amqpConnectAndPublish(Map<String, Object> endpoint) throws Exception {
// Configure SSL
SSLContext sslContext = createSSLContext();
// Create handler
SenderHandler handler = new SenderHandler(endpoint);
handler.setSslContext(sslContext);
// Create reactor
Reactor reactor = Proton.reactor(handler);
// Connect to host with SSL and SASL configuration
String host = (String) endpoint.get("host");
int port = ((Number) endpoint.get("port")).intValue();
// Use reactor's connection method with proper SSL/SASL setup
reactor.connectionToHost(host, port, handler);
// Run reactor
reactor.run();
}
function sendMessage(sender) {
// Increment message counter
messageCount++;
// Create dynamic message content with counter and timestamp
const now = new Date();
const timeString = now.toTimeString().split(' ')[0]; // HH:MM:SS format
const bodyText = `Hello World! Message #${messageCount} at ${timeString}`;
const bodyBinary = Buffer.from(bodyText, 'utf8');
// Parse application properties
const properties = JSON.parse(MESSAGE_APPLICATION_PROPERTIES_JSON);
// Create message with explicit data section for binary body
const message = {
// Use data_section to ensure body is sent as AMQP Data type (binary)
body: rhea.message.data_section(bodyBinary),
application_properties: properties
};
// Format properties in sorted order for consistent logging
const sortedProperties = JSON.stringify(properties, Object.keys(properties).sort());
console.log(`${new Date().toISOString()} INFO Sending message: body='${bodyText}', properties=${sortedProperties}`);
sender.send(message);
}
function amqpConnectAndPublish(endpoint) {
const container = rhea.create_container();
// Connection options with TLS configuration
const connectionOptions = {
host: endpoint.host,
port: parseInt(endpoint.port),
transport: 'tls',
// Client certificate authentication
key: fs.readFileSync(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM),
cert: fs.readFileSync(ACTOR_CERTIFICATE_CHAIN_AND_KEY_PEM),
// CA certificate
ca: [fs.readFileSync(CA_CERTIFICATE_PEM)],
// Connection settings
reconnect: false,
initial_reconnect_delay: 0,
max_reconnect_delay: 0,
idle_time_out: 5000, // 5 second heartbeat
container_id: 'javascript-delivery-example',
// Enable SASL EXTERNAL for client certificate authentication
enable_sasl_external: true
};
let sender = null;
let sendTimer = null;
container.on('connection_open', function (context) {
console.log(`${new Date().toISOString()} DEBUG Container reactor started`);
console.log(`${new Date().toISOString()} DEBUG Creating sender for target: ${endpoint.target}`);
// Create sender
sender = context.connection.open_sender({
target: {
address: endpoint.target
}
});
});
container.on('sendable', function (context) {
console.log(`${new Date().toISOString()} DEBUG Sender link opened, ready to send messages`);
// Send first message immediately (like Java onLinkRemoteOpen)
sendMessage(context.sender);
// Schedule subsequent messages every 1 second
sendTimer = setInterval(() => {
if (context.sender.sendable()) {
sendMessage(context.sender);
}
}, 1000);
});
container.on('accepted', function (context) {
// Message was accepted by the broker - schedule next message (like Python on_settled)
});
container.on('rejected', function (context) {
console.log(`${new Date().toISOString()} WARNING Message rejected:`, context.delivery.remote.error);
});
container.on('connection_close', function (context) {
console.log(`${new Date().toISOString()} INFO Connection closed`);
if (sendTimer) {
clearInterval(sendTimer);
}
});
container.on('connection_error', function (context) {
const error = context.connection.error || context.error;
console.log(`${new Date().toISOString()} SEVERE Connection error:`, error);
});
container.on('error', function (context) {
console.log(`${new Date().toISOString()} SEVERE Error:`, context.error);
});
container.on('disconnected', function (context) {
console.log(`${new Date().toISOString()} SEVERE Disconnected:`, context.error || 'Unknown error');
});
// Connect
console.log(`${new Date().toISOString()} INFO Connecting to amqps://${endpoint.host}:${endpoint.port}`);
container.connect(connectionOptions);
// Return a promise that runs indefinitely (like Python while thread.is_alive())
return new Promise(() => {
// This promise never resolves, keeping the connection alive
});
}
Create a delivery
delivery_create_response = api_create_delivery()
delivery_create_response_json = delivery_create_response.json();
var deliveryCreateResponse = await ApiCreateDeliveryAsync(client);
var deliveryCreateResponseContent = await deliveryCreateResponse.Content.ReadAsStringAsync();
var deliveryCreateResponseJson = JsonSerializer.Deserialize<DeliveryCreateResponse>(deliveryCreateResponseContent);
deliveryCreateResponse, err := apiCreateDelivery()
if err != nil {
log.Printf("Error creating delivery: %v", err)
return
}
deliveryCreateResponseJSON, err := parseResponseJSON(deliveryCreateResponse)
if err != nil {
log.Printf("Error parsing delivery create response: %v", err)
return
}
try (Response deliveryCreateResponse = apiCreateDelivery()) {
String responseBody = deliveryCreateResponse.body().string();
Map<String, Object> deliveryCreateResponseJson = objectMapper.readValue(responseBody,
new TypeReference<Map<String, Object>>() {});
const deliveryCreateResponse = await apiCreateDelivery();
const deliveryCreateResponseJson = deliveryCreateResponse.json();
Poll the delivery
delivery_id = delivery_create_response_json["id"]
delivery_status_response = api_get_delivery(delivery_id)
delivery_status_response_json = delivery_status_response.json()
log_json("Delivery %s status response" % delivery_id, delivery_status_response_json)
delivery_status = delivery_status_response_json["status"]
var deliveryId = deliveryCreateResponseJson.Id;
var deliveryStatusResponse = await ApiGetDeliveryAsync(deliveryId, client);
var deliveryStatusResponseContent = await deliveryStatusResponse.Content.ReadAsStringAsync();
var deliveryStatusResponseJson = JsonSerializer.Deserialize<DeliveryStatusResponse>(deliveryStatusResponseContent);
LogJson($"Delivery {deliveryId} status response", deliveryStatusResponseJson);
var deliveryStatus = deliveryStatusResponseJson.Status;
deliveryID := deliveryCreateResponseJSON["id"].(string)
deliveryStatusResponse, err := apiGetDelivery(deliveryID)
if err != nil {
log.Printf("Error getting delivery status: %v", err)
return
}
deliveryStatusResponseJSON, err := parseResponseJSON(deliveryStatusResponse)
if err != nil {
log.Printf("Error parsing delivery status response: %v", err)
return
}
logJSON(fmt.Sprintf("Delivery %s status response", deliveryID), deliveryStatusResponseJSON)
deliveryStatus := deliveryStatusResponseJSON["status"].(string)
String deliveryId = (String) deliveryCreateResponseJson.get("id");
Map<String, Object> deliveryStatusResponseJson;
String deliveryStatus;
try (Response deliveryStatusResponse = apiGetDelivery(deliveryId)) {
deliveryStatusResponseJson = objectMapper.readValue(deliveryStatusResponse.body().string(),
new TypeReference<Map<String, Object>>() {});
logJson("Delivery " + deliveryId + " status response", deliveryStatusResponseJson);
deliveryStatus = (String) deliveryStatusResponseJson.get("status");
}
const deliveryId = deliveryCreateResponseJson.id;
let deliveryStatusResponse = await apiGetDelivery(deliveryId);
let deliveryStatusResponseJson = deliveryStatusResponse.json();
logJson(`Delivery ${deliveryId} status response`, deliveryStatusResponseJson);
let deliveryStatus = deliveryStatusResponseJson.status;
Keep polling the delivery while REQUESTED
while delivery_status == "REQUESTED":
time.sleep(2)
delivery_status_response = api_get_delivery(delivery_id)
delivery_status_response_json = delivery_status_response.json()
delivery_status = delivery_status_response_json["status"]
log_json("Delivery %s status response" % delivery_id, delivery_status_response_json)
while (deliveryStatus == "REQUESTED")
{
await Task.Delay(2000);
deliveryStatusResponse = await ApiGetDeliveryAsync(deliveryId, client);
deliveryStatusResponseContent = await deliveryStatusResponse.Content.ReadAsStringAsync();
deliveryStatusResponseJson = JsonSerializer.Deserialize<DeliveryStatusResponse>(deliveryStatusResponseContent);
deliveryStatus = deliveryStatusResponseJson.Status;
}
LogJson($"Delivery {deliveryId} status response", deliveryStatusResponseJson);
for deliveryStatus == "REQUESTED" {
time.Sleep(2 * time.Second)
deliveryStatusResponse, err = apiGetDelivery(deliveryID)
if err != nil {
log.Printf("Error polling delivery status: %v", err)
return
}
deliveryStatusResponseJSON, err = parseResponseJSON(deliveryStatusResponse)
if err != nil {
log.Printf("Error parsing delivery status response: %v", err)
return
}
deliveryStatus = deliveryStatusResponseJSON["status"].(string)
}
logJSON(fmt.Sprintf("Delivery %s status response", deliveryID), deliveryStatusResponseJSON)
while ("REQUESTED".equals(deliveryStatus)) {
Thread.sleep(2000);
try (Response deliveryStatusResponse = apiGetDelivery(deliveryId)) {
deliveryStatusResponseJson = objectMapper.readValue(deliveryStatusResponse.body().string(),
new TypeReference<Map<String, Object>>() {});
deliveryStatus = (String) deliveryStatusResponseJson.get("status");
}
}
logJson("Delivery " + deliveryId + " status response", deliveryStatusResponseJson);
while (deliveryStatus === 'REQUESTED') {
await new Promise(resolve => setTimeout(resolve, 2000)); // Sleep 2 seconds
deliveryStatusResponse = await apiGetDelivery(deliveryId);
deliveryStatusResponseJson = deliveryStatusResponse.json();
deliveryStatus = deliveryStatusResponseJson.status;
}
logJson(`Delivery ${deliveryId} status response`, deliveryStatusResponseJson);
Use the endpoint information to connect
if delivery_status == "CREATED":
# NOTE to keep things simple, this code assumes that this response contains exactly one endpoint!
endpoint = delivery_status_response_json["endpoints"][0]
logging.info("Using endpoint %s" % endpoint)
amqp_connect_and_publish(endpoint)
if (deliveryStatus == "CREATED")
{
// NOTE to keep things simple, this code assumes that this response contains exactly one endpoint!
var endpoint = deliveryStatusResponseJson.Endpoints[0];
LogInfo($"Using endpoint {JsonSerializer.Serialize(endpoint)}");
await AmqpConnectAndPublishAsync(endpoint);
}
if deliveryStatus == "CREATED" {
// NOTE to keep things simple, this code assumes that this response contains exactly one endpoint!
endpoints := deliveryStatusResponseJSON["endpoints"].([]interface{})
endpoint := endpoints[0].(map[string]interface{})
log.Printf("Using endpoint %v", endpoint)
err = amqpConnectAndPublish(endpoint)
if err != nil {
log.Printf("Error in AMQP connection: %v", err)
}
} else {
if ("CREATED".equals(deliveryStatus)) {
// NOTE to keep things simple, this code assumes that this response contains exactly one endpoint!
List<Map<String, Object>> endpoints = (List<Map<String, Object>>) deliveryStatusResponseJson.get("endpoints");
Map<String, Object> endpoint = endpoints.get(0);
logger.info("Using endpoint " + endpoint);
amqpConnectAndPublish(endpoint);
}
if (deliveryStatus === 'CREATED') {
// NOTE to keep things simple, this code assumes that this response contains exactly one endpoint!
const endpoint = deliveryStatusResponseJson.endpoints[0];
console.log(`${new Date().toISOString()} INFO Using endpoint ${JSON.stringify(endpoint)}`);
await amqpConnectAndPublish(endpoint);
} else {
Full examples
| Language | Location | Description |
|---|---|---|
| Python | examples/delivery/python | Complete workflow: CREATE → POLL → CONNECT → USE |
| .NET | examples/delivery/dotnet | Complete workflow: CREATE → POLL → CONNECT → USE |
| Go | examples/delivery/go | Complete workflow: CREATE → POLL → CONNECT → USE |
| Java | examples/delivery/java | Complete workflow: CREATE → POLL → CONNECT → USE |
| JavaScript | examples/delivery/javascript | Complete workflow: CREATE → POLL → CONNECT → USE |