Issue on deleting stocks from adapter

rvkvino

I have some issue on my adapter. But my server code working properly. Some times adapter showing old data in user page. Some times it comes and disappear, and in some case now my server shows only 2 rows but adapter having 3 stock rows and displaying 3 rows in user page. But extra one row is too old data and deleted from server also. Even though it showing in adapter. If I restart the LS it working fine, but after some times again this issue comes and I need to restart the server. I couldn't find the reason and issue for this.
I'm using the following code,

private final HashMap<String, RateRow> liveratesnew = new HashMap<String, RateRow>();
private final HashMap<String, RateRow> liveratesold = new HashMap<String, RateRow>();

synchronized (liveratesnew) {
if ((liveratesnew.size() != liveratesold.size())) {
if (liveratesnew.size() > liveratesold.size()) {
synchronized (liveratesold) {
liveratesnew.forEach((k, v) -> {
if (!liveratesold.containsKey(liveratesnew.get(k).getItemName())) {
liveratesold.put(liveratesnew.get(k).getItemName(), v);
final HashMap<String, String> liverateevent = new HashMap<String, String>();
liverateevent.put("key", liveratesnew.get(k).getItemName());
liverateevent.put("desc", liveratesnew.get(k).getInstrumentName());
liverateevent.put("bid", liveratesnew.get(k).getBidRate());
liverateevent.put("ask", liveratesnew.get(k).getAskRate());
listener.onActualStatus(liveratesnew.get(k).getItemName(), liverateevent,
false, 1);
}
});
}
} else if (liveratesnew.size() < liveratesold.size()) {
final HashMap<String, String> removeoldkey = new HashMap<String, String>();
synchronized (liveratesold) {
for (Entry<String, RateRow> e : liveratesold.entrySet()) {
String key = e.getKey();
if (!liveratesnew.containsKey(liveratesold.get(key))) {
listener.onDeleteStatus(liveratesold.get(key).getItemName());
removeoldkey.put(key, liveratesold.get(key).getItemName());
// liveratesold.remove(liveratesold.get(key).getItemName());

}
}
}

for (Entry<String, String> e : removeoldkey.entrySet()) {
String key = e.getKey();
// listener.onDeleteStatus(key);
liveratesold.remove(key);
}
removeoldkey.clear();

}
}
}

Gianluca Finocchiaro

Hi rvkvino,

it's very hard to understand what is going on from the code you shown, as it its not clear where it is located inside your adapter.
That said, at first glance it seems that you are running against some race conditions, as you are checking liveratesold.size() outside the synchronized blocks on which it could be modified.

Try to move size checks inside synchronized blocks and get back to us.

Gianluca

rvkvino

public class ExternalFeedSimulator {
private static final Timer dispatcher = new Timer();
private final String REMOTE_RATE_FEED_URL = "serverurl";
private final ArrayList<RateRow> liverates = new ArrayList<RateRow>();
private final HashMap<String, RateRow> liveratesnew = new HashMap<String, RateRow>();
private final HashMap<String, RateRow> liveratesold = new HashMap<String, RateRow>();
private ExternalFeedListener listener;

public void start() {
sendGet();
// long waitTime = displayrates.computeNextWaitTime();
long waitTime = 500;
liveratescheduleGenerator(liverates, waitTime);
}

// HTTP GET request
public void sendGet() {
String url = REMOTE_RATE_FEED_URL;
URL obj;
try {
obj = new URL(url);
HttpURLConnection con = (HttpURLConnection) obj.openConnection();

// optional default is GET
con.setRequestMethod("GET");

// add request header
con.setRequestProperty("User-Agent", USER_AGENT);

// int responseCode = con.getResponseCode();

// System.out.println("Response Code : " + responseCode);

BufferedReader in = new BufferedReader(new InputStreamReader(con.getInputStream()));
String inputLine;
StringBuffer response = new StringBuffer();

while ((inputLine = in.readLine()) != null) {
response.append(inputLine);
}
in.close();

if (response.toString() == null || response.toString().equals(""))
return;
String strlowerresponse = response.toString();

String trade_type = getItemValue(strlowerresponse, "trade_type");
String rate_display = getItemValue(strlowerresponse, "rate_display");
String update_time = getItemValue(strlowerresponse, "gold_updatetime");
String market_closed_msg = "";
String commodityupdatetime = getItemValue(strlowerresponse, "commodityupdatetime");

trade_type = trade_type.trim();
int startIndex = 0;
liveratesnew.clear();
while (startIndex != -1) {
int item_start_pos = strlowerresponse.indexOf("<Commodity>", startIndex) + 11;
if (item_start_pos < 11)
break;
int item_end_pos = strlowerresponse.indexOf("</Commodity>", item_start_pos + 1);
if (item_end_pos <= -1)
break;
String strnode = strlowerresponse.substring(item_start_pos, item_end_pos);
String instrid = getItemValue(strnode, "id");
String instrname = getItemValue(strnode, "name");
String strsellrate = getItemValue(strnode, "selling_rate");
String strbuyrate = getItemValue(strnode, "buying_rate");
String strhighrate = getItemValue(strnode, "selling_high");
String strlowrate = getItemValue(strnode, "selling_low");

int ordernumber = Integer.parseInt(getItemValue(strnode, "ordernumber"));
if (strbuyrate.equalsIgnoreCase("") || strbuyrate.equalsIgnoreCase("0"))
strbuyrate = "-";
if (strsellrate.equalsIgnoreCase("") || strsellrate.equalsIgnoreCase("0"))
strsellrate = "-";
RateRow commrate = new RateRow("Item" + instrid, instrname, strbuyrate, strsellrate, strlowrate,
strhighrate, ordernumber);

liverates.add(commrate);
liveratesnew.put("Item" + instrid, commrate);
startIndex = item_end_pos;
}

if (trade_type.equalsIgnoreCase("3") || rate_display.equalsIgnoreCase("1")) {
rate_display = "0";
} else {
rate_display = "1";
}
RateRow marketstatus = new RateRow("Marketstatus", rate_display, commodityupdatetime, update_time,
market_closed_msg);
curmarketstatus.add(marketstatus);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (NumberFormatException e) {
System.out.println("not a number");
}
}

public synchronized void removeListener() {
// remove the listener
this.listener = null;
}

public String getItemValue(String strdata, String query) {
String strlowerdata = strdata.toLowerCase();
query = query.toLowerCase();
int item_start_pos = strlowerdata.indexOf("<" + query + ">") + query.length() + 2;
if (item_start_pos < query.length() + 2)
return "";
int item_end_pos = strlowerdata.indexOf("</" + query + ">");
if (item_end_pos < query.length() + 2)
return "";

String item_value = strdata.substring(item_start_pos, item_end_pos);
item_value = item_value.trim();
return item_value;
}

private void currentmarketstatusscheduleGenerator(final ArrayList<RateRow> nmarketstatus, long waitTime) {
dispatcher.schedule(new TimerTask() {
@Override
public void run() {
long nextWaitTime;
synchronized (nmarketstatus) {
if (listener != null) {
for (RateRow marketstatus : nmarketstatus) {
final HashMap<String, String> markstatus = new HashMap<String, String>();
markstatus.put("desc", marketstatus.getInstrumentName());
markstatus.put("ratedisplay", marketstatus.getBidRate());
markstatus.put("comupdatetime", marketstatus.getAskRate());
markstatus.put("updatetime", marketstatus.getLow_price());
markstatus.put("msg", marketstatus.getHigh_price());
listener.onEvent(marketstatus.getInstrumentName(), markstatus, false);
}
}
nmarketstatus.clear();
nextWaitTime = 500;
}

currentmarketstatusscheduleGenerator(curmarketstatus, nextWaitTime);
}
}, waitTime);
}
private void liveratescheduleGenerator(final ArrayList<RateRow> displayrates, long waitTime) {
dispatcher.schedule(new TimerTask() {
@Override
public void run() {
long nextWaitTime;
synchronized (displayrates) {
if (listener != null) {
for (RateRow curbidaskrates : displayrates) {
final HashMap<String, String> bidaskevent = new HashMap<String, String>();
bidaskevent.put("key", curbidaskrates.getItemName());
bidaskevent.put("desc", curbidaskrates.getInstrumentName());
bidaskevent.put("bid", curbidaskrates.getBidRate());
bidaskevent.put("ask", curbidaskrates.getAskRate());
bidaskevent.put("low", curbidaskrates.getLow_price());
bidaskevent.put("high", curbidaskrates.getHigh_price());
bidaskevent.put("order", Integer.toString(curbidaskrates.getDisplayorder()));
listener.onActualStatus(curbidaskrates.getItemName(), bidaskevent, false, 2);
// listener.onEvent(curbidaskrates.getItemName(),
// bidaskevent, false);
}

}
displayrates.clear();
nextWaitTime = 500;

synchronized (liveratesnew) {
if ((liveratesnew.size() != liveratesold.size())) {
if (liveratesnew.size() > liveratesold.size()) {
synchronized (liveratesold) {
liveratesnew.forEach((k, v) -> {
if (!liveratesold.containsKey(liveratesnew.get(k).getItemName())) {
liveratesold.put(liveratesnew.get(k).getItemName(), v);
final HashMap<String, String> liverateevent = new HashMap<String, String>();
liverateevent.put("key", liveratesnew.get(k).getItemName());
liverateevent.put("desc", liveratesnew.get(k).getInstrumentName());
liverateevent.put("bid", liveratesnew.get(k).getBidRate());
liverateevent.put("ask", liveratesnew.get(k).getAskRate());
liverateevent.put("low", liveratesnew.get(k).getLow_price());
liverateevent.put("high", liveratesnew.get(k).getHigh_price());
liverateevent.put("order",
Integer.toString(liveratesnew.get(k).getDisplayorder()));
listener.onActualStatus(liveratesnew.get(k).getItemName(), liverateevent,
false, 1);
}
});
}
} else if (liveratesnew.size() < liveratesold.size()) {

final HashMap<String, String> removeoldkey = new HashMap<String, String>();
synchronized (liveratesold) {
for (Entry<String, RateRow> e : liveratesold.entrySet()) {
String key = e.getKey();
if (!liveratesnew.containsKey(liveratesold.get(key))) {
listener.onDeleteStatus(liveratesold.get(key).getItemName());
removeoldkey.put(key, liveratesold.get(key).getItemName());
// liveratesold.remove(liveratesold.get(key).getItemName());

}
}
}

for (Entry<String, String> e : removeoldkey.entrySet()) {
String key = e.getKey();
// listener.onDeleteStatus(key);
liveratesold.remove(key);
}
removeoldkey.clear();

}
}
}

}
sendGet();
liveratescheduleGenerator(liverates, nextWaitTime);
}
}, waitTime);
}

public void setFeedListener(ExternalFeedListener listener) {
this.listener = listener;
}

public void sendRates(String itemName) {
liveratesnew.forEach((k, v) -> {
final RateRow liverate = liveratesnew.get(k);
if (liverate.getItemName().contains(itemName)) {
dispatcher.schedule(new TimerTask() {
@Override
public void run() {
synchronized (liverate) {
final HashMap<String, String> crates = new HashMap<String, String>();
crates.put("key", liverate.getItemName());
crates.put("desc", liverate.getInstrumentName());
crates.put("bid", liverate.getBidRate());
crates.put("ask", liverate.getAskRate());
crates.put("low", liverate.getLow_price());
crates.put("high", liverate.getHigh_price());
crates.put("order", Integer.toString(liverate.getDisplayorder()));
listener.onActualStatus(liverate.getItemName(), crates, true, 1);
}
}
}, 0);
}

});
}
}

Gianluca Finocchiaro

Hi rvkvino,

please try to remove possible race conditions as pointed out earlier; after that, we will evaluate whether to proceed with code inspection.

Thanks and Regards,
Gianluca

rvkvino

As You have mentioned I have checked the code many times and I have return the size checking code inside the synchronized blocks only. This issue not happening all the times. Some of the times only this issue happening and I need to restart the LS server then only get it work properly.

Gianluca Finocchiaro

Ok rkvino,

we need the LS server log to investigate the issue. Before that, please update lighstreamer_log_conf.xml as follows:

<logger name="LightstreamerLogger.subscriptions" level="DEBUG"/>
<logger name="LightstreamerLogger.pump" level="DEBUG"/>

and try to reproduce the problem.

Then, send the log file to support@lightstreamer.com.

Most important, please precisely describe what you are expecting in terms of rows to be displayed, providing us with a simple example of real data.

Thanks
Gianluca

rvkvino

We are facing this issue often. My remote adapter produces only 2 rows of stocks. Example Stock-A bid-123 ask-124 and Stock-B bid 122 ask 125 like this. But I'm receiving some times Stock-X with this 2 stocks and totally showing 3 stocks. This Stock-X was created previously and deleted. Even now not in a database also. While facing this issue I need to restart the LS server. If I restart LS server then it disappears Stock-X but after some times it again shows in the stock list. But this Stock-X now deleted from the server. Now I feel like this stock stored in LS server cookies or cache. If it stores in LS server in-memory means how do I delete all the in memory data and reproduce again.

Dario Crivelli

Since you talk about "rows" I suppose that you are referring to an item in COMMAND mode which contains the stocks as keys.
Can you please confirm that?

If you have an item in COMMAND mode and you send a DELETE event for a key, and then you see the key appearing again upon new subscriptions, this is unexpected.
However, only by watching at a concrete log, like instructed by Gianluca, can we find what is wrong.

One possibility is that you send a DELETE event for the key but immediately later, because of lack of synchronization, you send an UPDATE event for the same key. In this case, the Server would fix the inconsistency by converting the UPDATE into a new ADD. You can detect the case in the log, by finding a WARN like this:
Unexpected UPDATE event for key XXX. Event propagated as an ADD command.

Anyway, if this happens and you need to clean up the whole item, you can also try invoking clearSnapshot from your Data Adapter.

rvkvino

I'm using the Command mode only in my adapter, and also I have used delete command when removing stock from server. Now we are having only 2 stocks but it shows 3 stocks unfortunately. When I restart the LS server then it shows 2 stocks. How to clean up the whole item from the server.

Dario Crivelli

It's not clear to me what you mean by "using the Command mode only in my adapter"; I suppose that the clients also subscribe to your item by specifying COMMAND mode, am I right?
Hence, the second part of my previous answer applies.

rvkvino

Yes you are right, Client connected server by using COMMAND mode.
I have a code like below
public void clearStatus() {
synchronized (subscribedItems) {
Set<String> keys = subscribedItems.keySet();
for (String itemName : keys) {
listener.clearSnapshot(itemName);
}
}
}

In feedsimulator.java class I have code like below,
final HashMap<String, String> removeoldkey = new HashMap<String, String>();
synchronized (liveratesold) {
for (Entry<String, RateRow> e : liveratesold.entrySet()) {
String key = e.getKey();
if (!liveratesnew.containsKey(liveratesold.get(key))) {
listener.onDeleteStatus(liveratesold.get(key).getItemName());
removeoldkey.put(key, liveratesold.get(key).getItemName());
// liveratesold.remove(liveratesold.get(key).getItemName());

}
}
}
May i call clearStatus feedsimulator class.

rvkvino

Hi,We are facing the issue on COMMAND mode stock deleting. In my case, we are creating a dynamic stock list. So we are operating the stock list dynamically by add, delete and update. This everything works and add, update and delete working while we check on the server in both LS and our remote server and client page also getting reflect.But some times my client page displaying some extra stock details(It already deleted from our remote application). We have confirmed from our server no data available for this stock. But it showing in LS server and client page. We feel it persist in somewhere in LS server even after we deleted key like listener.onDeleteStatus(key);
For example, we are having 3 stocks like Stock-1, Stock-2, and Stock-3
Stock-1 Bid-2767 Ask-2769 High-2772 Low-2765
Stock-2 Bid-2767 Ask-2769 High-2772 Low-2765
Stock-3 Bid-2767 Ask-2769 High-2772 Low-2765
We have deleted Stock-3 from our server. In this time it getting deleted from LS and also client page COMMAND changed and displaying only Stock-1 and Stock-2.
But unfortunately, sometimes showing this Stock-3 on client page. We don't have any source on our server for this Stock-3.
And also we need to know is there any cache or in-memory storage for Ls. If there is in-memory storage can we clear daily or if we need at a time.

Dario Crivelli

Hello rkvino,

We received your log.

The log contains a lot of activity, even if in just 3 minutes.
So, we also need some coordinates to find the cited episode in the log.
Please specify the name of the item in COMMAND mode involved in the issue, as there are many in the log.
If possible, please also specify the name of the key that was supposed to be deleted and appeared again.

What I can say at this stage is that if I do a case-insensitive search for "delete" in the log I can see it only twice.
In practice, it is the value associated to a custom field, "FeedStatus", which is received from the Data Adapter and then forwarded to the client.
By the way, I also see the values "ADD" and "UPDATE" associated to this field.

This raises the suspect that you are not leveraging the items in COMMAND mode in the correct way.

Let's call "List" the item in COMMAND mode which contains the list of stocks.
Let's suppose that you would like to delete "Stock-3".
You have to:

create an update event (for instance a Map);
set field "key" to "Stock-3";
set field "command" to "delete";
send the update event for item "List" to Lightstreamer, by invoking listener.update(or listener.smartUpdate ).

In your code, you invoke some listener.onDeleteStatus, but this is not a method from LS APIs, this is still your code.
So, you should ensure that your onDeleteStatus behaves like I described above.
You can see what our Portfolio Demo Adapter does, as a reference.

-------------

If you add a key to a COMMAND item, it is kept in the item's state.
This means that other clients subscribing to this item will receive the key as part of the initial snapshot.
This is the only "in-memory storage" that we apply.
But if you delete the key in the proper way, it will be removed from the item state and no other caching will apply.

The Item's state is also cleared when the item is unsubscribed from by all clients; in this case, unsubscribe will also be invoked to your Data Adapter.
Upon the next client subscription, subscribe will be invoked to your Data Adapter and the item's state will be reconstructed from scratch.

A third possibility, as already suggested, is that you invoke clearSnapshot.
Did you try it?
The code shown seems correct, but I can't find occurrences of "clearSnapshot" in the log.
Obiously, you can use clearSnapshot only to ensure that no key will be present in the list.

rvkvino

Fir listener.onDeleteStatus I have return code like below in my DataAdapter

@Override
public void onDeleteStatus(String itemName) {
SubscriptionInfo si;
synchronized (subscribedItems) {
si = new SubscriptionInfo(new Boolean(false), new Boolean(true));
subscribedItems.remove(itemName, si);
onDelete(this.handle, itemName);
}
}

Dario Crivelli

The code shown is still internal to your Data Adapter and does not reference anything in Lightstreamer Adapter interface.
I mean that SubscriptionInfo, subscribedItems, and onDelete are still defined in your code.
So, it is difficult for us to devise any needed change.
This is furtherly complicated by the fact that the code is inspired in part to our StockQuotesDataAdapter.javacode sample (which defines SubscriptionInfo and subscribedItems) and in part by the PortfolioDataAdapter.javacode sample (which defines onDelete ).

Anyway, if you manage to invoke onDeleteand the implementation of onDelete is the one included in our PortfolioDataAdapter.java, this should correctly send the DELETE command.
However, we didn't see that reported in the log.
So, please debug your Adapter to ensure that

onDelete is invoked,
onDeleteinvokes listener.smartUpdate with the correct parameters,
after the invocation, the Server logs the update as expected (on LightstreamerLogger.subscriptions at DEBUG level).

If you dump the parameters upon the invocation of listener.smartUpdate, together with the current time, we can help you in comparing those with the log.

rvkvino

Hi,
I have updated code to clear snapshot while I delete the stock the below code will execute. After I cleared the snapshot still the deleted commodity displaying in client screen. Is there any other storage available in LS.

public void clearStatus() {
synchronized (subscribedItems) {
Set<String> keys = subscribedItems.keySet();
for (String itemName : keys) {
listener.clearSnapshot(itemName);
}
}
}

@Override
public void onDeleteStatus(String itemName) {
SubscriptionInfo si;
synchronized (subscribedItems) {
si = new SubscriptionInfo(new Boolean(false), new Boolean(true));
subscribedItems.remove(itemName, si);
onDelete(this.handle, itemName);
clearStatus();
}
}

Giuseppe Corti

Hi rvkvino,

I am not sure that the piece of code you posted, can work since the clearStatus method is called after the itemName is removed from the subscribedItems map.
Maybe there is some confusion between the COMMAND table keys and the ItemName representing the whole table.
While DELETE commands are executed on individual row keys, clearStatus must specify the name of the Item of entire table, the one you received with subscribe call.

Regards,
Giuseppe