Developing and running applications that use the DPS Toolkit

To create applications that use the DPS Toolkit, you'll need to:

1. Configure the product environment variables by entering the following command:

   
     source product-installation-root-directory/7.2.0.0/bin/streamsprofile.sh
   

2. Ensure that all of the following additional RPMs required by the DPS toolkit are present on your system:

   • curl
   • curl-devel
   • lua
   • lua-devel
   • openssl-devel
   • openldap-devel
3. Install and configure an external key-value data store that is supported by DPS.
4. Configure the DPS toolkit to connect to the data store from step 3. See the Data store connection configuration page for more info.

Note regarding fusing Java operators: Fusing two or more Java operators that utilize the Java DPS API requires that each operator have the same class path and the @SharedLoader annotation in its class definition. For example:

PrimitiveOperator(name="MyJavaOperator", namespace="com.teracloud.demo",  description="Java Operator MyJavaOperator")
@SharedLoader(true)
public class MyJavaOperator extends AbstractOperator {
  ...
}

rstring dummyRstring = "";
uint32 dummyUint32 = 0u;
mutable uint64 err = 0ul;
mutable uint64 dbStore_handle = 0ul;
dbStore_handle = dpsCreateStore("myDBStore1", dummyRstring, dummyUint32, err);

if (err == 0ul ) { //no error occurred
  //create lock for the store
  mutable uint64 lock_id = dlCreateOrGetLock("My db store lock", err);
  // Acquire the newly created lock, specifying a lease time and maximum time to wait to acquire the lock.
  float64 max_wait = 10.0;
  float64 lease_time = 10.0;
  dlAcquireLock(lock_id, lease_time, max_wait, err);
  //add a key/value pair to the store
  mutable boolean result = true;
  rstring key = "StreamsKey";
  uint32 value = 399;
  err = 0ul;
  result = dpsPut(dbStore_handle, key, value, err);
	
  if (err != 0ul) {
    //use  dpsGetLastStoreErrorCode() and  dpsGetLastStoreErrorString() as needed
  }
  // finished our store operations, release the lock
  err = 0ul;
  dlReleaseLock(lock_id, err);	
}

StoreFactory sf = DistributedStores.getStoreFactory();
Store store = null;

try {
  //specify the SPL types for the keys and values in the store
  String keyType = "rstring";
  String valueType = "int32";
  store = sf.createOrGetStore("Java Test Store1", keyType, valueType);
} catch (StoreFactoryException sfe) {
  // use	sfe.getErrorCode() and  sfe.getErrorMessage()) for more info
}

...
//once ready to access the store,
//get the lock for the store, may have previously been created
LockFactory lf = DistributedLocks.getLockFactory(); 
Lock myLock = lf.createOrGetLock("Lock_For_Test_Store1");

// Acquire the lock
try {
  myLock.acquireLock();
} catch (LockException le) {
  System.out.print("Unable to acquire the lock named 'Lock_For_Test_Store1'");
  System.out.println(" Error = " + le.getErrorCode() + ", Error msg = " + le.getErrorMessage());
  throw le;
}

//perform store operations
store.put("Key", 39);
store.put("Key2", 50);
//release the lock  when finished
myLock.releaseLock();

In addition, there are functions that can be called when an error occurs to return the last error code and a message describing the error. The correct function to call after an operation depends on the operation that was last executed.

• For errors relating to the store functions, use dpsGetLastStoreErrorCode() and dpsGetLastStoreErrorString().
• For errors related to the TTL functions, use dpsGetLastErrorCodeTTL() and dpsGetLastErrorStringTTL().
• For locking related errors, use dlGetLastDistributedLockErrorCode() and dlGetLastDistributedLockErrorString().

mutable uint64 err = 0ul;
mutable uint64 lock_id = dlCreateOrGetLock("My Sentinel Lock1", err);

if (err != 0ul) {
  rstring msg = dlGetLastDistributedLockErrorString();
  uint64 rc = dlGetLastDistributedLockErrorCode();
  printStringLn("Error creating lock, rc = " + (rstring)(rc) + ", msg =" + msg );
}