Why do I miss some prices randomly on multiple streams?

[jason ho 22]

Please look at my example: I am unsure why it decides to update sometimes and not for marketid 48,49,50… You can fire up two consoles to see the behaviour.

Any ideas?

//private static readonly Uri RPC_URI = new Uri("https://ciapipreprod.cityindextest9.co.uk/tradingapi");
        private static readonly Uri RPC_URI = new Uri("http://ciapipreprod.cityindextest9.co.uk/tradingapi");
        //private static readonly Uri STREAMING_URI = new Uri("https://pushpreprod.cityindextest9.co.uk/CITYINDEXSTREAMING");
        private static readonly Uri STREAMING_URI = new Uri("http://pushpreprod.cityindextest9.co.uk/CITYINDEXSTREAMING");
        private const string USERNAME = "DM032299";
        private const string PASSWORD = "password";
        public static StreamingClient.IStreamingClient _streamingClient = null;
        public static CIAPI.Rpc.Client _ctx = null;

        static void Main(string[] args)
        {
            //A single streaming client (connection) allows listening to many streams (channels)
            _ctx = new CIAPI.Rpc.Client(RPC_URI);
            _ctx.LogIn(USERNAME, PASSWORD);

            _streamingClient =
                StreamingClientFactory.CreateStreamingClient(
                    STREAMING_URI,
                    USERNAME,               /* Note how we use the same username */
                    _ctx.SessionId);  /* and and shared sessionId to connect */

            var gate = new ManualResetEvent(false);
            var isConnected = false;
            //The streaming client raises events when things happen
            _streamingClient.StatusChanged += (object sender, StatusEventArgs e) =>
            {
                if (e.Status.Contains("Connection established"))
                {
                    Console.WriteLine("*** CONNECTED ***");
                    isConnected = true;
                    gate.Set();
                }
            };
            //You have to connect the client to the server
            _streamingClient.Connect();

            Task[] streamingtasks = new Task[]  
            {  
                //Task.Factory.StartNew( () => TestAddMultipleStreams("154297")),                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481147")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481148")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481149")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481150")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481151")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481153")),                                                

            };
            //Task.WaitAll(streamingtasks);

            while (true)
            {
                Thread.Sleep(1000);
                Console.WriteLine("Waiting for updates...");
            }
            String stop = Console.ReadLine();
        }
        public static void TestAddMultipleStreams(String marketid)
        {
            StreamingClient.IStreamingListener<PriceDTO> listener = _streamingClient.BuildListener<PriceDTO>("PRICES.PRICE."+marketid);
            listener.MessageRecieved += (sender, message) =>
                {
                    Console.WriteLine(message.Data.MarketId + " " + message.Data.Bid + "," + message.Data.Offer + ",direction=" + message.Data.Direction);
                };

            listener.Start();

        }

[Guest mrdavidlaing]

MarketIds aren’t necessarily sequential; so its likely that 48,49,50… don’t exist, or aren’t currently active.

Since market prices are driven by underlying market activity at the exchanges; they don’t price regularly (and in fact don’t price at all when the market is closed).

[Guest mrdavidlaing]

 

Please look at my example: I am unsure why it decides to update sometimes and not for marketid 48,49,50… You can fire up two consoles to see the behaviour.

Any ideas?

//private static readonly Uri RPC_URI = new Uri("https://ciapipreprod.cityindextest9.co.uk/tradingapi");
        private static readonly Uri RPC_URI = new Uri("http://ciapipreprod.cityindextest9.co.uk/tradingapi");
        //private static readonly Uri STREAMING_URI = new Uri("https://pushpreprod.cityindextest9.co.uk/CITYINDEXSTREAMING");
        private static readonly Uri STREAMING_URI = new Uri("http://pushpreprod.cityindextest9.co.uk/CITYINDEXSTREAMING");
        private const string USERNAME = "DM032299";
        private const string PASSWORD = "password";
        public static StreamingClient.IStreamingClient _streamingClient = null;
        public static CIAPI.Rpc.Client _ctx = null;

        static void Main(string[] args)
        {
            //A single streaming client (connection) allows listening to many streams (channels)
            _ctx = new CIAPI.Rpc.Client(RPC_URI);
            _ctx.LogIn(USERNAME, PASSWORD);

            _streamingClient =
                StreamingClientFactory.CreateStreamingClient(
                    STREAMING_URI,
                    USERNAME,               /* Note how we use the same username */
                    _ctx.SessionId);  /* and and shared sessionId to connect */

            var gate = new ManualResetEvent(false);
            var isConnected = false;
            //The streaming client raises events when things happen
            _streamingClient.StatusChanged += (object sender, StatusEventArgs e) =>
            {
                if (e.Status.Contains("Connection established"))
                {
                    Console.WriteLine("*** CONNECTED ***");
                    isConnected = true;
                    gate.Set();
                }
            };
            //You have to connect the client to the server
            _streamingClient.Connect();

            Task[] streamingtasks = new Task[]  
            {  
                //Task.Factory.StartNew( () => TestAddMultipleStreams("154297")),                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481147")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481148")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481149")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481150")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481151")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481153")),                                                

            };
            //Task.WaitAll(streamingtasks);

            while (true)
            {
                Thread.Sleep(1000);
                Console.WriteLine("Waiting for updates...");
            }
            String stop = Console.ReadLine();
        }
        public static void TestAddMultipleStreams(String marketid)
        {
            StreamingClient.IStreamingListener<PriceDTO> listener = _streamingClient.BuildListener<PriceDTO>("PRICES.PRICE."+marketid);
            listener.MessageRecieved += (sender, message) =>
                {
                    Console.WriteLine(message.Data.MarketId + " " + message.Data.Bid + "," + message.Data.Offer + ",direction=" + message.Data.Direction);
                };

            listener.Start();

        }

Could you give some more detail on the 2 console scenario – perhaps some log output? Are you saying that one of the consoles will get a price update and the other won’t?

[jason ho 22]

 

Please look at my example: I am unsure why it decides to update sometimes and not for marketid 48,49,50… You can fire up two consoles to see the behaviour.

Any ideas?

//private static readonly Uri RPC_URI = new Uri("https://ciapipreprod.cityindextest9.co.uk/tradingapi");
        private static readonly Uri RPC_URI = new Uri("http://ciapipreprod.cityindextest9.co.uk/tradingapi");
        //private static readonly Uri STREAMING_URI = new Uri("https://pushpreprod.cityindextest9.co.uk/CITYINDEXSTREAMING");
        private static readonly Uri STREAMING_URI = new Uri("http://pushpreprod.cityindextest9.co.uk/CITYINDEXSTREAMING");
        private const string USERNAME = "DM032299";
        private const string PASSWORD = "password";
        public static StreamingClient.IStreamingClient _streamingClient = null;
        public static CIAPI.Rpc.Client _ctx = null;

        static void Main(string[] args)
        {
            //A single streaming client (connection) allows listening to many streams (channels)
            _ctx = new CIAPI.Rpc.Client(RPC_URI);
            _ctx.LogIn(USERNAME, PASSWORD);

            _streamingClient =
                StreamingClientFactory.CreateStreamingClient(
                    STREAMING_URI,
                    USERNAME,               /* Note how we use the same username */
                    _ctx.SessionId);  /* and and shared sessionId to connect */

            var gate = new ManualResetEvent(false);
            var isConnected = false;
            //The streaming client raises events when things happen
            _streamingClient.StatusChanged += (object sender, StatusEventArgs e) =>
            {
                if (e.Status.Contains("Connection established"))
                {
                    Console.WriteLine("*** CONNECTED ***");
                    isConnected = true;
                    gate.Set();
                }
            };
            //You have to connect the client to the server
            _streamingClient.Connect();

            Task[] streamingtasks = new Task[]  
            {  
                //Task.Factory.StartNew( () => TestAddMultipleStreams("154297")),                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481147")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481148")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481149")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481150")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481151")),                                
                Task.Factory.StartNew( () => TestAddMultipleStreams("400481153")),                                                

            };
            //Task.WaitAll(streamingtasks);

            while (true)
            {
                Thread.Sleep(1000);
                Console.WriteLine("Waiting for updates...");
            }
            String stop = Console.ReadLine();
        }
        public static void TestAddMultipleStreams(String marketid)
        {
            StreamingClient.IStreamingListener<PriceDTO> listener = _streamingClient.BuildListener<PriceDTO>("PRICES.PRICE."+marketid);
            listener.MessageRecieved += (sender, message) =>
                {
                    Console.WriteLine(message.Data.MarketId + " " + message.Data.Bid + "," + message.Data.Offer + ",direction=" + message.Data.Direction);
                };

            listener.Start();

        }

@mrdavidlaing

Hi I am trying to produce some log files to show you this example. What happens is I run the same console app at the same time. 1 console will get prices for say marketid 48 and receive live prices… the other which is subscribing to the same set of marketids will not receive the live prices for marketid 48… The behaviour of this is random when I launch the console, if I subscribe to the 4 markets sometimes it will subscribe to 48 sometimes it doesn’t it seems it will subscribe to 2 or 3 of the 4 but never all 4. The 2 or 3 is random again.

Currently I am only getting one message back on several marketids and no updates. Can you give me more information about the test server is there certain times which it will publish or stream prices or anyway I can check the test server is live and streaming.