Beanstalk Client

@library("beanstalk", "0.0.0");

This library provides an implementation of a Beanstalk client. The client allows you to interact with a Beanstalk server, sending and receiving messages in a variety of formats. The implementation respects the Beanstalk protocol : https://raw.githubusercontent.com/beanstalkd/beanstalkd/master/doc/protocol.txt

Type: Beanstalk

Represents the Beanstalk client.

Properties

• host: String: The host address of the Beanstalk server.
• port: int: The port number of the Beanstalk server.
• topic_name: String: The name of the Beanstalk tube (topic).
• format: BeanstalkFormat?: Optional. The format of the messages in the Beanstalk tube. Currently, only plain text is supported. (In the current version, this value will be ignored and only plain text is supported)
• buffer_file_path: String?: The path to the buffer file. This field is indispensable for the method read_all_to_buffer() to work. For the read() method, it is optional. If specified, messages will be persisted into the buffer file.
• on_msg: function? : Only for threaded mode. User defined callback function to be invoked by threaded background listeners.

Methods

Basic IO methods

• write(v: any): Send a message of any type to the Beanstalk server.
• read(): any?: If the buffer_file_path field is not specified, it reserves and returns a received message as BeanstalkJob. Otherwise, it reserves a message, persist the message to a buffer file and return the message’s ending offset in the buffer file.
• read_all(): Array<BeanstalkJob?>: Read up to 128 messages from the Beanstalk server. It returns the received messages as an array of BeanstalkJob.
• read_all_to_buffer(): int?: Read up to 128 messages from the Beanstalk server, persist them to a buffer file, and return the content ending’s offset in the buffer file.
• release_all(ids: Array<int>): Notify the Beanstalk server to release all messages in the ids array.
• commit_all(ids: Array<int>): Delete all messages in the ids array from the Beanstalk server.

Threaded mode configuration

This Beanstalk library also supports background threaded listeners. Each listener can be configured to watch different (or same) topics and servers. Each listener handles the incoming messages by invoking a user defined callback function. This callback function should always take one unique input argument which is a String. To configure such listeners, the static fn configure(config: Array<Beanstalk>); method should be called.

Be careful, threaded listeners only work in server mode. It means the greycat application has to be launched by the greycat serve command. Please check the Demo session to find a simple example.

Take care:

• When buffer_file_path is not specified:
  For no matter which reading method, the messages won’t be deleted after consumption. You have to delete them manually using commit_all() method.

  • The method read_all_to_buffer will not be usable.
  • The method read returns one <BeanstalkJob>.
  • The method read_all returns an Array of <BeanstalkJob>.

• When buffer_file_path is specified:
  Messages retrieved by read_all_to_buffer and read wil be deleted from the server since the messages would be persisted to the buffer_fie while consumption.

  • The methods read_all_to_buffer and read will persist the received message(s) into the indicated buffer file. These methods return one the end-offset of the received/persisted message(s) in the buffer file.
  • The method read_all returns an Array of <BeanstalkJob>.

• For read_all and read_all_to_buffer, the max number of message to read is limited by #GC_BEANSTALK_MESSAGE_IN_QUEUE_MAX

Connection management

For a blocking read operation, if the library fails to connect to the server, it will attempt to reconnect every 5 seconds for up to one hour. If the connection cannot be established within this period, it will terminate with an error message indicating the connection failure.

For threaded listeners, if the initial connection to the server fails or if the connection is lost at any point, the library will continuously attempt to reconnect. During the first hour after the disconnection, it will try to reconnect every 5 seconds. If the disconnection persists for more than an hour, the library will try to reconnect every 30 minutes.

Type: BeanstalkJob

Represents a job in the Beanstalk tube.

• id: int: The ID of the job.
• content: any: The content of the job.

Demo

Demo for regular single thread use case

@library("beanstalk", "0.0.0");



fn main() {
    var bst = Beanstalk{host: "127.0.0.1", port: 11300, topic_name: "project_echo_topic"};
    var msg_out = "This is your GreyCat speaking!";
    var msg_in;

    println("#### Echo example");
    println("## Outbound message: " + msg_out);

    // Write two messages and read_all to receive them. Finally, Deletes all messages using commit_all()
    bst.write(msg_out);
    bst.write(msg_out);
    msg_in = bst.read_all();
    println("## Inbound read_all message: " + "${msg_in}");
    var ids = Array<int>{};
    for (var i = 0; i < msg_in.size(); i++){
        ids.add(msg_in.get(i)!!.id);
    }
    bst.commit_all(ids);
    ids = null;

    // Write one message, receive -> release -> receive again -> delete the message using commit_all()
    bst.write(msg_out);
    msg_in = bst.read();
    println("## Inbound read message: " + "${msg_in}");

    ids = Array<int>{};
    ids.add(msg_in!!.id as int);

    bst.release_all(ids);
    msg_in = bst.read();
    println("## Inbound read again message: " + "${msg_in}");
    bst.commit_all(ids);
    ids = null;


    bst.write(msg_out);
    msg_in = bst.read();
    ids = Array<int>{};
    ids.add(msg_in!!.id as int);
    println("## Inbound read message: " + "${msg_in}");
    bst.commit_all(ids);

    if (msg_in!!.content == msg_out) {
        println("#### Echo test passed!\n");
    } else {
        println("#### Echo test failed!\n");
    }
}

Demo for threaded listener

@library("beanstalk", "0.0.0");




fn hello(s: String){
    println("## Inbound read message: " + "${s}");
}

fn main(){
    Beanstalk::configure([Beanstalk{host: "127.0.0.1", port: 11300, topic_name: "project_echo_topic", on_msg: project::hello}]);
}