Using Memorystore in Apama

Forum
1

Hi,

I am trying to use memorystore in apama to store tables in the inmemorry storage. I am getting an error in the MemorySore.mon file.
The error is

Error on line 844 in action prepare in event com.apama.memorystore.Store: PluginException - Null chunk provided to MemoryStore plugin (in plugin method StoreChunk_prepare3) - C:\SoftwareAG\Apama\monitors\data_storage\MemoryStore.mon

My code where i am getting exception is line 31


		action onload() {

		Schema schema := new Schema;
		schema.fields := ["y"];
		schema.types := [];
		integer storeId := storage.prepareInMemory("storeInMem"); 
		Finished f; 
		on Finished(storeId,*,*):f 
		   onStorePrepared(f);
		DataViewAddDefinition add := new DataViewAddDefinition;
		add.dvName := DataViewName;
		add.dvDisplayName := "Datasets";
		add.fieldNames := ["y"];
		//add.fieldTypes := ["integer","string","float","float","float","float"];
		add.fieldTypes := ["float"];
		add.keyFields := ["y"];
		add.msgId := integer.getUnique().toString();

		integer integerXVariable := 0; 
		while integerXVariable <8 {

			integerXVariable := integerXVariable+1;
			schema.fields.append("X"+integerXVariable.toString());
			schema.types.append("float");
			add.fieldTypes.append("float");
			add.fieldNames.append("X"+integerXVariable.toString());
		}

		schema.exposeMemoryView := true;
		schema.memoryViewDisplayName :="MemoryDataView";
		integer tableId := store.prepare("tableInMem", schema); 
		on Finished(tableId,*,*):f onTablePrepared(f); 
		
		route add;



		DataViewException dvException;

		on DataViewException(msgId=add.msgId):dvException
		   and not DataViewDefinition(msgId = add.msgId) {
			log "Exception: " + dvException.message;

		}

		on DataViewDefinition(msgId = add.msgId) {
			Dataset ds;
			on all Dataset():ds{

				float fx := 8.0;
				float fy := 13.0;


				float fx1 := 0.0;
				float fy1 := -33960.0;

				float fx2 := 52483.0;
				float fy2 := 33965.0;

				float dx := ((fx2-fx1)/fx);
				float dy := (fy2-fy1)/fy;

				float ix := (ds.x1)/dx;
				integer floorx := ix.floor();

				integer integerXVariable := 0; 
				sequence<float> s := [];

				while integerXVariable <8 {

					if floorx = integerXVariable then{
						s.append(1.0);
					}
					else {
						s.append(0.0);
					}
					integerXVariable := integerXVariable+1;
				}

				Row row := tbl.get(ds.y1.toString());

				if not row.inTable()
				then{
					row.setFloat("X1",s[0]);
					row.setFloat("X2",s[1]);
					row.setFloat("X3",s[2]);
					row.setFloat("X4",s[3]);
					row.setFloat("X5",s[4]);
					row.setFloat("X6",s[5]);
					row.setFloat("X7",s[6]);
					row.setFloat("X8",s[7]);
					row.commit();
				}
				else
				{	
					row.setFloat("X1",s[0]+row.getFloat("X1"));
					row.setFloat("X2",s[1]+row.getFloat("X2"));
					row.setFloat("X3",s[2]+row.getFloat("X3"));
					row.setFloat("X4",s[3]+row.getFloat("X4"));
					row.setFloat("X5",s[4]+row.getFloat("X5"));
					row.setFloat("X6",s[5]+row.getFloat("X6"));
					row.setFloat("X7",s[6]+row.getFloat("X7"));
					row.setFloat("X8",s[7]+row.getFloat("X8"));
					row.commit();
				}

				print " Definition DS -- " + ds.toString() + "sequence ---" + s.toString();
				DataViewAddItem item := new DataViewAddItem;
				item.dvName := DataViewName;
				item.owner := "*";
				item.fieldValues := [ds.y1.toString(),s[0].toString(),s[1].toString(),s[2].toString(),s[3].toString(),s[4].toString(),s[5].toString(),s[6].toString(),s[7].toString()];
				item.timeStamp :=currentTime;
				item.msgId := integer.getUnique().toString();

				route item;

				DataViewItemException itemExp;
				on DataViewItemException(msgId = item.msgId):itemExp
				   and not DataViewItem(msgId = item.msgId){

					log "Exception: "+itemExp.message;				
				}

				DataViewItem added;
				on DataViewItem(msgId = item.msgId):added {
					log("DataViewItem: "+ added.dvItemId.toString());
					spawn updateItem(ds.y1,s);

				}

			}

		}
	}
2

Hi,

Have you opened a store and set the ‘store’ variable? Something like:

store := storage.open(“MyStore”);

possibly in onStorePrepared(). Can you copy in the onStorePrepared() action?

Thanks,
Martin

3

Hi,

Thanks for the reply. Please find the code below.



action onStorePrepared(Finished f) { 
		if not f.success then { log "Store Whoops"; die; } 
		store := storage.open("storeInMem");
	}
	action onTablePrepared(Finished f) { 
		if not f.success then { log "Table Whoops"; die; } 
		tbl := store.open("tableInMem");
	}
4

The problem is that the ‘onStorePrepared()’ (which sets the store variable) isn’t called until the ‘Finished()’ event is sent back from the plugin. But processing will continue into the DataView section and the line ‘integer tableId := store.prepare(“tableInMem”, schema);’ before store has been set.

You need to move all the code after the first ‘on Finished()’ into a block so it’s only processed once the Finished() event has been sent back:-




	Schema schema := new Schema;
	schema.fields := ["y"];
	schema.types := [];
	integer storeId := storage.prepareInMemory("storeInMem"); 
	Finished f; 
	on Finished(storeId,*,*):f {
		onStorePrepared(f);

		DataViewAddDefinition add := new DataViewAddDefinition;
		add.dvName := DataViewName;
		add.dvDisplayName := "Datasets";
		add.fieldNames := ["y"];
		//add.fieldTypes := ["integer","string","float","float","float","float"];
		add.fieldTypes := ["float"];
		add.keyFields := ["y"];
		add.msgId := integer.getUnique().toString();

		integer integerXVariable := 0; 
		while integerXVariable <8 {

			integerXVariable := integerXVariable+1;
			schema.fields.append("X"+integerXVariable.toString());
			schema.types.append("float");
			add.fieldTypes.append("float");
			add.fieldNames.append("X"+integerXVariable.toString());
		}

		schema.exposeMemoryView := true;
		schema.memoryViewDisplayName :="MemoryDataView";
		integer tableId := store.prepare("tableInMem", schema); 
		on Finished(tableId,*,*):f onTablePrepared(f); 
		
		route add;


		DataViewException dvException;

		on DataViewException(msgId=add.msgId):dvException
		   and not DataViewDefinition(msgId = add.msgId) {
			log "Exception: " + dvException.message;

		}

		on DataViewDefinition(msgId = add.msgId) {
			Dataset ds;
			on all Dataset():ds{

				float fx := 8.0;
				float fy := 13.0;


				float fx1 := 0.0;
				float fy1 := -33960.0;

				float fx2 := 52483.0;
				float fy2 := 33965.0;

				float dx := ((fx2-fx1)/fx);
				float dy := (fy2-fy1)/fy;

				float ix := (ds.x1)/dx;
				integer floorx := ix.floor();

				integer integerXVariable := 0; 
				sequence<float> s := [];

				while integerXVariable <8 {

					if floorx = integerXVariable then{
						s.append(1.0);
					}
					else {
						s.append(0.0);
					}
					integerXVariable := integerXVariable+1;
				}

				Row row := tbl.get(ds.y1.toString());

				if not row.inTable()
				then{
					row.setFloat("X1",s[0]);
					row.setFloat("X2",s[1]);
					row.setFloat("X3",s[2]);
					row.setFloat("X4",s[3]);
					row.setFloat("X5",s[4]);
					row.setFloat("X6",s[5]);
					row.setFloat("X7",s[6]);
					row.setFloat("X8",s[7]);
					row.commit();
				}
				else
				{	
					row.setFloat("X1",s[0]+row.getFloat("X1"));
					row.setFloat("X2",s[1]+row.getFloat("X2"));
					row.setFloat("X3",s[2]+row.getFloat("X3"));
					row.setFloat("X4",s[3]+row.getFloat("X4"));
					row.setFloat("X5",s[4]+row.getFloat("X5"));
					row.setFloat("X6",s[5]+row.getFloat("X6"));
					row.setFloat("X7",s[6]+row.getFloat("X7"));
					row.setFloat("X8",s[7]+row.getFloat("X8"));
					row.commit();
				}

				print " Definition DS -- " + ds.toString() + "sequence ---" + s.toString();
				DataViewAddItem item := new DataViewAddItem;
				item.dvName := DataViewName;
				item.owner := "*";
				item.fieldValues := [ds.y1.toString(),s[0].toString(),s[1].toString(),s[2].toString(),s[3].toString(),s[4].toString(),s[5].toString(),s[6].toString(),s[7].toString()];
				item.timeStamp :=currentTime;
				item.msgId := integer.getUnique().toString();

				route item;

				DataViewItemException itemExp;
				on DataViewItemException(msgId = item.msgId):itemExp
				   and not DataViewItem(msgId = item.msgId){

					log "Exception: "+itemExp.message;				
				}

				DataViewItem added;
				on DataViewItem(msgId = item.msgId):added {
					log("DataViewItem: "+ added.dvItemId.toString());
					spawn updateItem(ds.y1,s);

				}
			}
		}
	}
}