vb.net - OnUpdate - how to 'listen' to the data stream?

lcmtech

Hi all,

Hoping someone can point me in the right direction as I'm going round in circles with this now. I'm trying to write a vb.net client - I can connect to the server, I can subscribe to a table, but I cannot for the life of me work out how to trigger something when one of the subscribed items changes. The code I've got so far is below;

Dim lsClient = New Lightstreamer.DotNet.Client.LSClient
Dim lsClientConn = New Lightstreamer.DotNet.Client.ConnectionInfo
Dim lsClientListenHandy As Lightstreamer.DotNet.Client.IHandyTableListener = Nothing

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

lsClient.OpenConnection(lsClientConn, lsClientListenHandy)

'fields, group, items, schema

Dim arritems() = {"item1", "item2", "item3", "item4", "item5", "item6", "item7", "item8", "item9", "item10", "item11", "item12", "item13", "item14", "item15"}
Dim arrFields() = {"last_price", "time", "pct_change", "bid_quantity", "bid", "ask", "ask_quantity", "min", "max", "ref_price", "open_price", "stock_name", "item_status"}
Dim lsExtendedTable = New Lightstreamer.DotNet.Client.ExtendedTableInfo(arritems, "MERGE", arrFields, 0)
lsExtendedTable.DataAdapter = "QUOTE_ADAPTER"
lsClient.SubscribeTable(lsExtendedTable, lsClientListenHandy, True)

And I also have a class declared for the lsClientListenHandy, which I thought was the key but I'm clearly missing something.

Public Class Listen
Implements IHandyTableListener
Public Shared sResult As String
Sub OnUpdate(itemPos As Integer, itemName As String, update As Lightstreamer.DotNet.Client.IUpdateInfo) Implements IHandyTableListener.OnUpdate
sResult = update.ItemPos.ToString & " - " & update.ToString
Controller.UpdateRec(sResult)

Throw New NotImplementedException()
End Sub
Public Sub OnSnapshotEnd(itemPos As Integer, itemName As String) Implements IHandyTableListener.OnSnapshotEnd
Throw New NotImplementedException()
End Sub
Public Sub OnRawUpdatesLost(itemPos As Integer, itemName As String, lostUpdates As Integer) Implements IHandyTableListener.OnRawUpdatesLost
Throw New NotImplementedException()
End Sub
Public Sub OnUnsubscr(itemPos As Integer, itemName As String) Implements IHandyTableListener.OnUnsubscr
Throw New NotImplementedException()
End Sub
Public Sub OnUnsubscrAll() Implements IHandyTableListener.OnUnsubscrAll
Throw New NotImplementedException()
End Sub
End Class

What am I missing to get the OnUpdate to do anything in code? (I've got a test lightstreamer server set up and I can see it sending updates in the dashboard.

Thanks in advance!

Giuseppe Corti

Hi lcmtech,

First of all, we should be sure that the subscription is managed by the server properly.
Please could you check the server log (lightstreamer.log) for any error or warn message?
You can also attach a snippet of the log here.

Regards,
Giuseppe

lcmtech

Hi Giuseppe,

Thanks for the response. Logs from a connection attempt are as follows - looks ok to my untrained eye. The connection closed message is when I stop the program from running.

7-Apr-18 16:20:45,907|INFO |L.requests |SERVER POOLED THREAD 4 |Serving request: /lightstreamer/create_session.txt --> LS_op2=create&LS_cid=jqWtj1tg4pkpW37AL3O4hwLri8L4NCc&LS_adapter_set=FULLPORTFOLIODEMO&LS_content_length=50000000&LS_report_info=true on "Lightstreamer HTTP Server" from 192.168.1.131:57902
07-Apr-18 16:20:45,912|INFO |L.requests |SERVER POOLED THREAD 4 |Starting new session: S3e20a41e5537260fM8fcT2045911 (dotnet_client 2.1.2 build 1015) on "Lightstreamer HTTP Server" from 192.168.1.131:57902
07-Apr-18 16:20:45,945|INFO |L.requests |SERVER POOLED THREAD 4 |Serving request: /lightstreamer/control.txt --> LS_session=S3e20a41e5537260fM8fcT2045911&LS_op=add&LS_table=1&LS_id=item1%20item2%20item3%20item4%20item5%20item6%20item7%20item8%20item9%20item10%20item11%20item12%20item13%20item14%20item15&LS_mode=MERGE&LS_schema=last_price%20time%20pct_change%20bid_quantity%20bid%20ask%20ask_quantity%20min%20max%20ref_price%20open_price%20stock_name%20item_status&LS_data_adapter=QUOTE_ADAPTER on "Lightstreamer HTTP Server" from 192.168.1.131:57903
07-Apr-18 16:20:45,945|INFO |L.requests |SERVER POOLED THREAD 4 |Controlling session: S3e20a41e5537260fM8fcT2045911 on "Lightstreamer HTTP Server" from 192.168.1.131:57903
07-Apr-18 16:21:06,426|INFO |streamerMonitorText|Timer-0 |Total threads = 66, Total heap = 95617024 (free = 69170296), Sessions = 3 (max = 3), New sessions = [+1, -0], Connections = 4 (max = 9), New connections = [+2, -0], In-pool threads = 22, Active threads = 0, Available threads = 22, Queued tasks = 0, Pool queue wait = 0, NIO write queue = 0, NIO write queue wait = 0, NIO write selectors = 1, NIO total selectors = 4, Subscribed items = 20 (for 27 subscriptions), Inbound throughput = 5 updates/s (pre-filtered = 5), Outbound throughput = 5.49 updates/s (4.89 kbit/s, max = 9.15), Lost updates = 0 (total = 0), Total bytes sent = 382064, Client messages throughput = 0 msgs/s (0 kbit/s, max = 0), Total messages handled = 0, Extra sleep = 0, Notify delay = 0
07-Apr-18 16:21:08,375|INFO |L.requests |PUMP POOLED THREAD 1 |Closed session S3e20a41e5537260fM8fcT2045911 with internal cause code: 38/5 (Connection interrupted by client).
07-Apr-18 16:23:06,871|INFO |streamerMonitorText|Timer-0 |Total threads = 66, Total heap = 95617024 (free = 61512656), Sessions = 2 (max = 3), New sessions = [+0, -1], Connections = 2 (max = 9), New connections = [+0, -2], In-pool threads = 22, Active threads = 0, Available threads = 22, Queued tasks = 0, Pool queue wait = 0, NIO write queue = 0, NIO write queue wait = 0, NIO write selectors = 1, NIO total selectors = 4, Subscribed items = 5 (for 12 subscriptions), Inbound throughput = 0.5 updates/s (pre-filtered = 0.5), Outbound throughput = 0.99 updates/s (0.94 kbit/s, max = 9.15), Lost updates = 0 (total = 0), Total bytes sent = 389584, Client messages throughput = 0 msgs/s (0 kbit/s, max = 0), Total messages handled = 0, Extra sleep = 0, Notify delay = 0
07-Apr-18 16:25:07,295|INFO |streamerMonitorText|Timer-0 |Total threads = 66, Total heap = 95617024 (free = 54038512), Sessions = 2 (max = 3), New sessions = [+0, -0], Connections = 2 (max = 9), New connections = [+0, -0], In-pool threads = 22, Active threads = 0, Available threads = 22, Queued tasks = 0, Pool queue wait = 0, NIO write queue = 0, NIO write queue wait = 0, NIO write selectors = 1, NIO total selectors = 4, Subscribed items = 5 (for 12 subscriptions), Inbound throughput = 0.5 updates/s (pre-filtered = 0.5), Outbound throughput = 0.99 updates/s (0.95 kbit/s, max = 9.15), Lost updates = 0 (total = 0), Total bytes sent = 396456, Client messages throughput = 0 msgs/s (0 kbit/s, max = 0), Total messages handled = 0, Extra sleep = 0, Notify delay = 0
07-Apr-18 16:27:07,823|INFO |streamerMonitorText|Timer-0 |Total threads = 67, Total heap = 95617024 (free = 72987440), Sessions = 2 (max = 3), New sessions = [+0, -0], Connections = 2 (max = 9), New connections = [+0, -0], In-pool threads = 22, Active threads = 0, Available threads = 22, Queued tasks = 0, Pool queue wait = 0, NIO write queue = 0, NIO write queue wait = 0, NIO write selectors = 1, NIO total selectors = 4, Subscribed items = 5 (for 12 subscriptions), Inbound throughput = 0.5 updates/s (pre-filtered = 0.5), Outbound throughput = 1 updates/s (0.87 kbit/s, max = 9.15), Lost updates = 0 (total = 0), Total bytes sent = 403546, Client messages throughput = 0 msgs/s (0 kbit/s, max = 0), Total messages handled = 0, Extra sleep = 0, Notify delay = 0
07-Apr-18 16:29:08,190|INFO |streamerMonitorText|Timer-0 |Total threads = 66, Total heap = 95617024 (free = 65416824), Sessions = 2 (max = 3), New sessions = [+0, -0], Connections = 2 (max = 9), New connections = [+0, -0], In-pool threads = 22, Active threads = 0, Available threads = 22, Queued tasks = 0, Pool queue wait = 0, NIO write queue = 0, NIO write queue wait = 0, NIO write selectors = 1, NIO total selectors = 4, Subscribed items = 5 (for 12 subscriptions), Inbound throughput = 0.5 updates/s (pre-filtered = 0.5), Outbound throughput = 1 updates/s (0.9 kbit/s, max = 9.15), Lost updates = 0 (total = 0), Total bytes sent = 410522, Client messages throughput = 0 msgs/s (0 kbit/s, max = 0), Total messages handled = 0, Extra sleep = 0, Notify delay = 0
07-Apr-18 16:31:08,654|INFO |streamerMonitorText|Timer-0 |Total threads = 66, Total heap = 95617024 (free = 58006488), Sessions = 2 (max = 3), New sessions = [+0, -0], Connections = 2 (max = 9), New connections = [+0, -0], In-pool threads = 22, Active threads = 0, Available threads = 22, Queued tasks = 0, Pool queue wait = 0, NIO write queue = 0, NIO write queue wait = 0, NIO write selectors = 1, NIO total selectors = 4, Subscribed items = 5 (for 12 subscriptions), Inbound throughput = 0.5 updates/s (pre-filtered = 0.5), Outbound throughput = 1 updates/s (0.83 kbit/s, max = 9.15), Lost updates = 0 (total = 0), Total bytes sent = 417384, Client messages throughput = 0 msgs/s (0 kbit/s, max = 0), Total messages handled = 0, Extra sleep = 0, Notify delay = 0
07-Apr-18 16:33:09,247|INFO |streamerMonitorText|Timer-0 |Total threads = 66, Total heap = 95617024 (free = 76763808), Sessions = 2 (max = 3), New sessions = [+0, -0], Connections = 2 (max = 9), New connections = [+0, -0], In-pool threads = 22, Active threads = 0, Available threads = 22, Queued tasks = 0, Pool queue wait = 0, NIO write queue = 0, NIO write queue wait = 0, NIO write selectors = 1, NIO total selectors = 4, Subscribed items = 5 (for 12 subscriptions), Inbound throughput = 0.49 updates/s (pre-filtered = 0.49), Outbound throughput = 0.99 updates/s (0.94 kbit/s, max = 9.15), Lost updates = 0 (total = 0), Total bytes sent = 424446, Client messages throughput = 0 msgs/s (0 kbit/s, max = 0), Total messages handled = 0, Extra sleep = 0, Notify delay = 0
07-Apr-18 16:35:09,592|INFO |streamerMonitorText|Timer-0 |Total threads = 66, Total heap = 95617024 (free = 69219088), Sessions = 2 (max = 3), New sessions = [+0, -0], Connections = 2 (max = 9), New connections = [+0, -0], In-pool threads = 22, Active threads = 0, Available threads = 22, Queued tasks = 0, Pool queue wait = 0, NIO write queue = 0, NIO write queue wait = 0, NIO write selectors = 1, NIO total selectors = 4, Subscribed items = 5 (for 12 subscriptions), Inbound throughput = 0.5 updates/s (pre-filtered = 0.5), Outbound throughput = 1 updates/s (0.79 kbit/s, max = 9.15), Lost updates = 0 (total = 0), Total bytes sent = 431280, Client messages throughput = 0 msgs/s (0 kbit/s, max = 0), Total messages handled = 0, Extra sleep = 0, Notify delay = 0
07-Apr-18 16:36:05,305|INFO |L.requests |SERVER POOLED THREAD 5 |Serving request: /lightstreamer/create_session.txt --> LS_op2=create&LS_cid=jqWtj1tg4pkpW37AL3O4hwLri8L4NCc&LS_adapter_set=FULLPORTFOLIODEMO&LS_content_length=50000000&LS_report_info=true on "Lightstreamer HTTP Server" from 192.168.1.131:58073
07-Apr-18 16:36:05,306|INFO |L.requests |SERVER POOLED THREAD 5 |Starting new session: S7a6355c76c631370M8fcT3605306 (dotnet_client 2.1.2 build 1015) on "Lightstreamer HTTP Server" from 192.168.1.131:58073
07-Apr-18 16:36:30,048|INFO |L.requests |SERVER POOLED THREAD 5 |Serving request: /lightstreamer/control.txt --> LS_session=S7a6355c76c631370M8fcT3605306&LS_op=add&LS_table=1&LS_id=item1%20item2%20item3%20item4%20item5%20item6%20item7%20item8%20item9%20item10%20item11%20item12%20item13%20item14%20item15&LS_mode=MERGE&LS_schema=last_price%20time%20pct_change%20bid_quantity%20bid%20ask%20ask_quantity%20min%20max%20ref_price%20open_price%20stock_name%20item_status&LS_data_adapter=QUOTE_ADAPTER on "Lightstreamer HTTP Server" from 192.168.1.131:58075
07-Apr-18 16:36:30,049|INFO |L.requests |SERVER POOLED THREAD 5 |Controlling session: S7a6355c76c631370M8fcT3605306 on "Lightstreamer HTTP Server" from 192.168.1.131:58075
07-Apr-18 16:37:00,217|INFO |L.requests |PUMP POOLED THREAD 1 |Closed session S7a6355c76c631370M8fcT3605306 with internal cause code: 38/5 (Connection interrupted by client).

Giuseppe Corti

Hi lcmtech,

From the server log it seems that the create session and subscribe requests were processed properly by the server.
Also from the 16:21:06 MonitorText log line it seems confirmed that the server were actually pushing some updates to the clients, the "Outbound throughput" figure is higher than normal values after the session closes.

So there are no reasons why the OnUpdate method of your client should not be called.

Please could you simply add this in your OnUpdate method (and I think that you could also remove the NotImplementedException)?

Console.WriteLine("New " + itemName & ":" & update.GetNewValue("last_price"))

And also jsut to have the finest details in the server log could you add these log settings (lightstreamer_log_conf.xml)?

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

This is something working for me:

[SYNTAX=VB] Dim items() = {"item1", "item2", "item3", "item4", "item5", "item6", "item7", "item8", "item9", "item10", "item11", "item12", "item13", "item14", "item15"}
Dim fields() = {"last_price", "time", "pct_change", "bid_quantity", "bid", "ask", "ask_quantity", "min", "max", "ref_price", "open_price", "stock_name", "item_status"}
Dim tableRef As SubscribedTableKey

Dim tableInfo As New ExtendedTableInfo(items, "MERGE", fields, True)

tableInfo.DataAdapter = "QUOTE_ADAPTER"

tableRef = lsc.SubscribeTable(tableInfo,
New LSSubscribeListener(upds),
False)[/SYNTAX]

[SYNTAX=Vb]Imports Lightstreamer.DotNet.Client

Public Class LSSubscribeListener
Implements Lightstreamer.DotNet.Client.IHandyTableListener

Dim updatesText As TextBox

Public Sub New(t As TextBox)
updatesText = t
End Sub

Private Delegate Sub AppendTextBoxDelegate(ByVal TB As TextBox, ByVal txt As String)

Private Sub AppendTextBox(ByVal TB As TextBox, ByVal txt As String)
If TB.InvokeRequired Then
TB.Invoke(New AppendTextBoxDelegate(AddressOf AppendTextBox), New Object() {TB, txt})
Else
TB.Text = txt
End If
End Sub

Public Sub OnUpdate(itemPos As Integer, itemName As String, update As IUpdateInfo) Implements IHandyTableListener.OnUpdate
Dim newMsg As String

Console.WriteLine("New " + itemName & ":" & update.GetNewValue("last_price"))
'updatesText.Text = itemName & ":" & update.GetNewValue("last_price")

newMsg = itemName & ":" & update.GetNewValue("last_price")
AppendTextBox(updatesText, newMsg)

Throw New NotImplementedException()
End Sub

Public Sub OnSnapshotEnd(itemPos As Integer, itemName As String) Implements IHandyTableListener.OnSnapshotEnd

End Sub

Public Sub OnRawUpdatesLost(itemPos As Integer, itemName As String, lostUpdates As Integer) Implements IHandyTableListener.OnRawUpdatesLost

End Sub

Public Sub OnUnsubscr(itemPos As Integer, itemName As String) Implements IHandyTableListener.OnUnsubscr

End Sub

Public Sub OnUnsubscrAll() Implements IHandyTableListener.OnUnsubscrAll

End Sub
End Class[/SYNTAX]

Regards,
Giuseppe

lcmtech

Hi Guiseppe,

Thank you for the response, that was incredibly useful! Made me realise I'd missed out the 'New' keyword in my subscription call, which was causing it not to update (line 10 in your first block of code). It's working superbly now.

Much appreciated!

lcmtech