Creating StreamBase Java Clients

For enqueue clients, the dialog presents options to enable:

In the generated code, you can also explicitly flush the buffer of a specified stream, or flush all buffers.

By enabling tuple buffering and experimenting with these parameters, you may be able to improve the efficiency and performance of your enqueue code. A single enqueue of a buffer containing (for example) 100 tuples should be more efficient than making 100 separate enqueue operations.

The example below demonstrates how to use this feature. The buffer size is set to 100 tuples and the buffer flush interval is set to 1000 milliseconds (one second).

package com.mycompany.mysbapps.clients;

import java.util.*;

import com.streambase.sb.*;
import com.streambase.sb.client.*;
   .
   .
   .
public class MyEnqueuerClient {

   private final static String SB_URI = "sb://localhost:10000";
   private static StreamBaseClient client = null;
   private String[] streamNames = null;
   private Collection tuples = null;
   private int buf_size = 100;
   private int flush_interval = 1000;
   .
   .
   .
   public MyEnqueuerClient(StreamBaseURI uri) throws StreamBaseException {
      System.out.println("Connecting to " + uri.toString() + "...");
      client = new StreamBaseClient(uri);
      System.out.println("Connected to " + uri.toString());

      // Turn on buffering. A WakeAndFlushBuffer thread is only
      // started if flush_interval > 0.
      client.enableBuffering(buf_size, flush_interval);
   }

In the StreamBase Client libraries, buffering is only turned on if the value for the buf_size parameter is greater than zero. (The buf_size parameter specifies the number of tuples, not a byte limit.) Also the buffer is flushed at a regular interval specified by the flush_interval parameter. If the flush_interval is not greater than zero, the buffer will only be flushed when it reaches capacity (it is filled with tuples).

If buffering is enabled and the flush_interval is set, the buffer will be flushed periodically based on the specified interval. However, to have more control over enqueuing buffered tuples, you can also use the flushBuffer(stream_name) method of the StreamBaseClient class to enqueue the contents of a buffer immediately. Or, you can use the flushAllBuffers() method to enqueue the contents of all buffers. These methods assume that the caller has a lock on _buffers_lock.

Generally you want to flush a buffer when you are concerned that the tuples in the buffer may become stale (staleness is relative to the particular application). Let's say, for example, your buffer size is 1000 and your flush_interval is 0 (this value turns off periodic flushing). The buffer will automatically be flushed when it is filled. That is, when the 1000th tuple has been enqueued. If 300 tuples have just been enqueued and it is unclear when or if more tuples will arrive to fill the buffer, it would be a good idea to call flushBuffer() or flushAllBuffers().