stomp

README

xk6-stomp

This is a Stomp protocol client library for k6, implemented as an extension using the xk6 system.

Build

To build a k6 binary with this extension, first ensure you have the prerequisites:

• Go toolchain
• Git

Then:

1. Install xk6:

go install go.k6.io/xk6/cmd/xk6@latest

1. Build the binary:

xk6 build --with github.com/walterwanderley/xk6-stomp

Example test script

1. Start a Stomp server (ActiveMQ, RabbitMQ, etc)

docker run -p 8161:8161 -p 61613:61613 rmohr/activemq

1. Write the test code

// test.js
import stomp from 'k6/x/stomp';

// connect to broker
const client = stomp.connect({
    addr: 'localhost:61613',
    timeout: '2s',
});

export default function () {
    // send a message to '/my/destination' with text/plain as MIME content-type
    client.send('my/destination', 'text/plain', 'Hello xk6-stomp!');

    const subscribeOpts = {
        ack: 'client' // client-individual or auto (default)
    }
    // subscribe to receive messages from 'my/destination' with the client ack mode
    const subscription = client.subscribe('my/destination', subscribeOpts); 

    // read the message
    const msg = subscription.read();

    // show the message as a string
    console.log('msg', msg.string());
    
    // ack the message
    client.ack(msg);
    
    // unsubscribe from destination
    subscription.unsubscribe();
}

export function teardown() {
    // disconnect from broker
    client.disconnect();
}

1. Result output:

$ ./k6 run _examples/test.js 

          /\      |‾‾| /‾‾/   /‾‾/   
     /\  /  \     |  |/  /   /  /    
    /  \/    \    |     (   /   ‾‾\  
   /          \   |  |\  \ |  (‾)  | 
  / __________ \  |__| \__\ \_____/ .io

  execution: local
     script: _examples/test.js
     output: -

  scenarios: (100.00%) 1 scenario, 1 max VUs, 10m30s max duration (incl. graceful stop):
           * default: 1 iterations for each of 1 VUs (maxDuration: 10m0s, gracefulStop: 30s)

INFO[0000] msg Hello xk6-stomp!                          source=console

running (00m00.0s), 0/1 VUs, 1 complete and 0 interrupted iterations
default ✓ [======================================] 1 VUs  00m00.0s/10m0s  1/1 iters, 1 per VU

     █ teardown

     data_received........: 322 B 16 kB/s
     data_sent............: 251 B 12 kB/s
     iteration_duration...: avg=5.23ms min=1.12ms med=5.23ms max=9.33ms p(90)=8.51ms p(95)=8.92ms
     iterations...........: 1     48.05613/s
     stomp_ack_count......: 1     48.05613/s
     stomp_read_count.....: 1     48.05613/s
     stomp_read_time......: avg=6.2ms  min=6.2ms  med=6.2ms  max=6.2ms  p(90)=6.2ms  p(95)=6.2ms 
     stomp_send_count.....: 1     48.05613/s
     stomp_send_time......: avg=5.5µs  min=5.5µs  med=5.5µs  max=5.5µs  p(90)=5.5µs  p(95)=5.5µs

Documentation

Index

• type Client
  • func (c *Client) Ack(m *Message) error
  • func (c *Client) Begin() *Transaction
  • func (c *Client) BeginWithError(ctx context.Context) (*Transaction, error)
  • func (c *Client) Connect(opts *Options) *Client
  • func (c *Client) Disconnect() error
  • func (c *Client) Nack(m *Message) error
  • func (c *Client) Send(destination, contentType string, body []byte, opts *SendOptions) (err error)
  • func (c *Client) Server() string
  • func (c *Client) Session() string
  • func (c *Client) Subscribe(destination string, opts *SubscribeOptions) (*Subscription, error)
• type Listener
• type Message
  • func (m *Message) JSON(selector ...string) goja.Value
  • func (m *Message) String() string
• type Options
• type RootModule
  • func New() *RootModule
  • func (*RootModule) NewModuleInstance(vu modules.VU) modules.Instance
• type SendOptions
• type StatsReadWriteClose
  • func (s *StatsReadWriteClose) Read(p []byte) (int, error)
  • func (s *StatsReadWriteClose) Write(p []byte) (int, error)
• type Stomp
  • func (s *Stomp) Exports() modules.Exports
• type SubscribeOptions
• type Subscription
  • func NewSubscription(client *Client, sc *stomp.Subscription, listener Listener) *Subscription
  • func (s *Subscription) Read() (msg *Message, err error)
  • func (s *Subscription) Unsubscribe(opts ...func(*frame.Frame) error) error
• type Transaction
  • func (tx *Transaction) Ack(m *Message) error
  • func (tx *Transaction) Nack(m *Message) error
  • func (tx *Transaction) Send(destination, contentType string, body []byte, opts *SendOptions) (err error)
• type VerboseReadWriteClose
  • func (v *VerboseReadWriteClose) Read(p []byte) (int, error)
  • func (v *VerboseReadWriteClose) Write(p []byte) (int, error)

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client is the Stomp conn wrapper.

func (*Client) Ack

func (c *Client) Ack(m *Message) error

Ack acknowledges a message received from the STOMP server.

func (*Client) Begin

func (c *Client) Begin() *Transaction

Begin is used to start a transaction.

func (*Client) BeginWithError

func (c *Client) BeginWithError(ctx context.Context) (*Transaction, error)

BeginWithError is used to start a transaction, but also returns the error.

func (*Client) Connect

func (c *Client) Connect(opts *Options) *Client

Connect to a stomp server

func (*Client) Disconnect

func (c *Client) Disconnect() error

Disconnect will disconnect from the STOMP server.

func (*Client) Nack

func (c *Client) Nack(m *Message) error

Nack indicates to the server that a message was not received by the client.

func (*Client) Send

func (c *Client) Send(destination, contentType string, body []byte, opts *SendOptions) (err error)

Send sends a message to the STOMP server.

func (*Client) Server

func (c *Client) Server() string

Server returns the STOMP server identification.

func (*Client) Session

func (c *Client) Session() string

Session returns the session identifier.

func (*Client) Subscribe

func (c *Client) Subscribe(destination string, opts *SubscribeOptions) (*Subscription, error)

Subscribe creates a subscription on the STOMP server.

type Listener

type Listener func(*Message)

Listener is a callback function to execute when the subscription reads a message

type Message

type Message struct {
	*stomp.Message
	// contains filtered or unexported fields
}

Message is a decorator to add string and json methods

func (*Message) JSON

func (m *Message) JSON(selector ...string) goja.Value

func (*Message) String

func (m *Message) String() string

type Options

type Options struct {
	Addr     string
	Protocol string
	Path     string
	Query    string
	Timeout  string
	TLS      bool
	Headers  map[string]string
	Host     string

	User string
	Pass string

	MessageSendTimeout string
	ReceiptTimeout     string

	Heartbeat struct {
		Incoming string
		Outgoing string
	}

	ReadBufferSize      int
	ReadChannelCapacity int

	WriteBufferSize      int
	WriteChannelCapacity int

	Verbose bool
}

type RootModule

type RootModule struct{}

func New

func New() *RootModule

func (*RootModule) NewModuleInstance

func (*RootModule) NewModuleInstance(vu modules.VU) modules.Instance

type SendOptions

type SendOptions struct {
	Headers map[string]string
	Receipt bool
}

type StatsReadWriteClose

type StatsReadWriteClose struct {
	io.ReadWriteCloser
	// contains filtered or unexported fields
}

func (*StatsReadWriteClose) Read

func (s *StatsReadWriteClose) Read(p []byte) (int, error)

func (*StatsReadWriteClose) Write

func (s *StatsReadWriteClose) Write(p []byte) (int, error)

type Stomp

type Stomp struct {
	*Client
	// contains filtered or unexported fields
}

Stomp is the k6 extension for a Stomp client.

func (*Stomp) Exports

func (s *Stomp) Exports() modules.Exports

type SubscribeOptions

type SubscribeOptions struct {
	Ack      string
	Headers  map[string]string
	Id       string
	Listener Listener
}

type Subscription

type Subscription struct {
	*stomp.Subscription
	// contains filtered or unexported fields
}

func NewSubscription

func NewSubscription(client *Client, sc *stomp.Subscription, listener Listener) *Subscription

func (*Subscription) Read

func (s *Subscription) Read() (msg *Message, err error)

func (*Subscription) Unsubscribe

func (s *Subscription) Unsubscribe(opts ...func(*frame.Frame) error) error

type Transaction

type Transaction struct {
	*stomp.Transaction
	// contains filtered or unexported fields
}

func (*Transaction) Ack

func (tx *Transaction) Ack(m *Message) error

func (*Transaction) Nack

func (tx *Transaction) Nack(m *Message) error

func (*Transaction) Send

func (tx *Transaction) Send(destination, contentType string, body []byte, opts *SendOptions) (err error)

type VerboseReadWriteClose

type VerboseReadWriteClose struct {
	io.ReadWriteCloser
}

func (*VerboseReadWriteClose) Read

func (v *VerboseReadWriteClose) Read(p []byte) (int, error)

func (*VerboseReadWriteClose) Write

func (v *VerboseReadWriteClose) Write(p []byte) (int, error)