+2

.gitignore

reviewed

··· 1 1 + /target 2 2 + /Cargo.lock

+26

Cargo.toml

reviewed

··· 1 1 + [package] 2 2 + authors = ["Kaushik Chakraborty <git@kaushikc.org>"] 3 3 + name = "kiteticker-async" 4 4 + version = "0.1.0" 5 5 + edition = "2021" 6 6 + license = "Apache-2.0" 7 7 + readme = "README.md" 8 8 + repository = "https://github.com/kaychaks/kiteticker-async" 9 9 + documentation = "https://docs.rs/kiteticker-async/0.1.0/kiteticker-async/" 10 10 + description = """ 11 11 + Async version of the ticker module of the kiteconnect-rs crate. 12 12 + """ 13 13 + categories = ["asynchronous", "finance"] 14 14 + keywords = ["ticker", "zerodha", "web-sockets", "trading", "real-time"] 15 15 + 16 16 + [dependencies] 17 17 + serde_json = "1.0" 18 18 + serde = { version = "1.0", features = ["derive"] } 19 19 + tokio = { version = "1.28.2", features = ["full"] } 20 20 + tokio-tungstenite = { version = "0.20.1", features = ["native-tls"] } 21 21 + futures-util = { version = "0.3.28", features = ["sink"] } 22 22 + tokio-stream = { version = "0.1.14", features = ["full"] } 23 23 + url = "2.4.1" 24 24 + 25 25 + [dev-dependencies] 26 26 + tokio = { version = "1", features = ["test-util"] }

+17

justfile

reviewed

··· 1 1 + cargo-doc-command := "cargo doc --no-deps" 2 2 + 3 3 + check: 4 4 + cargo check 5 5 + 6 6 + test: 7 7 + cargo test 8 8 + 9 9 + build: 10 10 + cargo build 11 11 + 12 12 + doc: 13 13 + rm -r target/doc 14 14 + {{cargo-doc-command}} 15 15 + doc-open: 16 16 + rm -r target/doc 17 17 + {{cargo-doc-command}} --open

+3

rustfmt.toml

reviewed

··· 1 1 + tab_spaces = 2 2 2 + max_width = 80 3 3 + use_field_init_shorthand = true

+30

src/lib.rs

reviewed

··· 1 1 + #![allow( 2 2 + clippy::cognitive_complexity, 3 3 + clippy::large_enum_variant, 4 4 + clippy::needless_doctest_main 5 5 + )] 6 6 + #![warn( 7 7 + missing_debug_implementations, 8 8 + rust_2018_idioms, 9 9 + unreachable_pub 10 10 + )] 11 11 + #![doc(test( 12 12 + no_crate_inject, 13 13 + attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables)) 14 14 + ))] 15 15 + 16 16 + //! Async implementation of Kite Connect's WebSocket Steaming API 17 17 + //! 18 18 + //! This crate provides types to subscribe and receive live quotes for instruments during market hours via WebSockets. 19 19 + //! The response is parsed and converted into Rust types. 20 20 + //! The WebSocket connection is managed by the library and reconnected automatically. 21 21 + //! 22 22 + 23 23 + mod models; 24 24 + pub use models::{ 25 25 + Depth, DepthItem, Exchange, Mode, Request, TextMessage, Tick, TickMessage, 26 26 + TickerMessage, OHLC, 27 27 + }; 28 28 + 29 29 + pub mod ticker; 30 30 + pub use ticker::{KiteTickerAsync, KiteTickerSubscriber};

+492

src/models.rs

reviewed

··· 1 1 + use serde::{Deserialize, Serialize}; 2 2 + use std::{ops::Div, time::Duration}; 3 3 + 4 4 + fn value(input: &[u8]) -> Option<u32> { 5 5 + let value = i32::from_be_bytes(input[0..=3].try_into().unwrap()); 6 6 + value.try_into().ok() 7 7 + } 8 8 + 9 9 + fn value_short(input: &[u8]) -> Option<u16> { 10 10 + let value = i16::from_be_bytes(input[0..=1].try_into().unwrap()); 11 11 + value.try_into().ok() 12 12 + } 13 13 + 14 14 + fn price(input: &[u8], exchange: &Exchange) -> Option<f64> { 15 15 + let value = i32::from_be_bytes(input[0..4].try_into().unwrap()) as f64; 16 16 + if exchange.divisor() > 0_f64 { 17 17 + Some(value.div(exchange.divisor())) 18 18 + } else { 19 19 + None 20 20 + } 21 21 + } 22 22 + 23 23 + pub(crate) fn packet_length(bs: &[u8]) -> usize { 24 24 + i16::from_be_bytes(bs[0..=1].try_into().unwrap()) as usize 25 25 + } 26 26 + 27 27 + #[derive(Debug, Clone, Default)] 28 28 + /// 29 29 + /// Quote packet structure 30 30 + /// 31 31 + pub struct Tick { 32 32 + pub mode: Mode, 33 33 + pub instrument_token: u32, 34 34 + pub exchange: Exchange, 35 35 + pub is_tradable: bool, 36 36 + pub is_index: bool, 37 37 + 38 38 + pub last_traded_qty: Option<u32>, 39 39 + pub avg_traded_price: Option<f64>, 40 40 + pub last_price: Option<f64>, 41 41 + pub volume_traded: Option<u32>, 42 42 + pub total_buy_qty: Option<u32>, 43 43 + pub total_sell_qty: Option<u32>, 44 44 + pub ohlc: Option<OHLC>, 45 45 + 46 46 + pub last_traded_timestamp: Option<Duration>, 47 47 + pub oi: Option<u32>, 48 48 + pub oi_day_high: Option<u32>, 49 49 + pub oi_day_low: Option<u32>, 50 50 + pub exchange_timestamp: Option<Duration>, 51 51 + 52 52 + pub net_change: Option<f64>, 53 53 + pub depth: Option<Depth>, 54 54 + } 55 55 + 56 56 + impl Tick { 57 57 + fn set_instrument_token(&mut self, input: &[u8]) -> &mut Self { 58 58 + self.instrument_token = value(&input[0..=3]).unwrap(); 59 59 + self.exchange = ((self.instrument_token & 0xFF) as usize).into(); 60 60 + self 61 61 + } 62 62 + 63 63 + fn set_change(&mut self) -> &mut Self { 64 64 + self.net_change = self 65 65 + .ohlc 66 66 + .as_ref() 67 67 + .map(|o| o.close) 68 68 + .map(|close_price| { 69 69 + if let Some(last_price) = self.last_price { 70 70 + if close_price == 0_f64 { 71 71 + return None; 72 72 + } else { 73 73 + Some(((last_price - close_price) * 100.0).div(close_price)) 74 74 + } 75 75 + } else { 76 76 + None 77 77 + } 78 78 + }) 79 79 + .unwrap_or_default(); 80 80 + self 81 81 + } 82 82 + } 83 83 + 84 84 + impl From<&[u8]> for Tick { 85 85 + fn from(input: &[u8]) -> Self { 86 86 + let mut tick = Tick::default(); 87 87 + 88 88 + let parse_ltp = |t: &mut Tick, i: &[u8]| { 89 89 + // 0 - 4 bytes : instrument token 90 90 + t.set_instrument_token(i); 91 91 + // 4 - 8 bytes : ltp 92 92 + if let Some(bs) = i.get(4..8) { 93 93 + t.mode = Mode::LTP; 94 94 + t.last_price = price(bs, &t.exchange); 95 95 + } 96 96 + }; 97 97 + 98 98 + let parse_quote = |t: &mut Tick, i: &[u8], is_index: bool| { 99 99 + if is_index { 100 100 + if let Some(bs) = i.get(8..28) { 101 101 + t.mode = Mode::Quote; 102 102 + // 8 - 24 bytes : ohlc 103 103 + t.ohlc = OHLC::from(&bs[0..16], &t.exchange); 104 104 + // 24 - 28 bytes : Price change 105 105 + // t.net_change = price(&bs[16..=19], &t.exchange); 106 106 + t.set_change(); 107 107 + } 108 108 + } else { 109 109 + if let Some(bs) = i.get(8..44) { 110 110 + t.mode = Mode::Quote; 111 111 + // 8 - 12 bytes : last traded quantity 112 112 + t.last_traded_qty = value(&bs[0..4]); 113 113 + // 12 - 16 bytes : avg traded price 114 114 + t.avg_traded_price = price(&bs[4..8], &t.exchange); 115 115 + // 16 - 20 bytes : volume traded today 116 116 + t.volume_traded = value(&bs[8..12]); 117 117 + // 20 - 24 bytes : total buy quantity 118 118 + t.total_buy_qty = value(&bs[12..16]); 119 119 + // 24 - 28 bytes : total sell quantity 120 120 + t.total_sell_qty = value(&bs[16..20]); 121 121 + // 28 - 44 bytes : ohlc 122 122 + t.ohlc = OHLC::from(&bs[20..36], &t.exchange); 123 123 + 124 124 + t.set_change(); 125 125 + } 126 126 + } 127 127 + }; 128 128 + 129 129 + let parse_full = |t: &mut Tick, i: &[u8], is_index: bool| { 130 130 + if is_index { 131 131 + if let Some(bs) = i.get(28..32) { 132 132 + t.mode = Mode::Full; 133 133 + // 28 - 32 bytes : exchange time 134 134 + t.exchange_timestamp = 135 135 + value(bs).map(|x| Duration::from_secs(x.into())); 136 136 + } 137 137 + } else { 138 138 + if let Some(bs) = i.get(44..184) { 139 139 + t.mode = Mode::Full; 140 140 + // 44 - 48 bytes : last traded timestamp 141 141 + t.last_traded_timestamp = 142 142 + value(&bs[0..4]).map(|x| Duration::from_secs(x.into())); 143 143 + 144 144 + // 48 - 52 bytes : oi 145 145 + t.oi = value(&bs[4..8]); 146 146 + // 52 - 56 bytes : oi day high 147 147 + t.oi_day_high = value(&bs[8..12]); 148 148 + // 56 - 60 bytes : oi day low 149 149 + t.oi_day_low = value(&bs[12..16]); 150 150 + // 60 - 64 bytes : exchange time 151 151 + t.exchange_timestamp = 152 152 + value(&bs[16..20]).map(|x| Duration::from_secs(x.into())); 153 153 + // 64 - 184 bytes : market depth 154 154 + t.depth = Depth::from(&bs[20..140], &t.exchange); 155 155 + } 156 156 + } 157 157 + }; 158 158 + 159 159 + parse_ltp(&mut tick, input); 160 160 + if !tick.exchange.is_tradable() { 161 161 + tick.is_index = true; 162 162 + tick.is_tradable = false; 163 163 + 164 164 + parse_quote(&mut tick, input, true); 165 165 + parse_full(&mut tick, input, true); 166 166 + } else { 167 167 + tick.is_index = false; 168 168 + tick.is_tradable = true; 169 169 + 170 170 + parse_quote(&mut tick, input, false); 171 171 + parse_full(&mut tick, input, false); 172 172 + } 173 173 + 174 174 + tick 175 175 + } 176 176 + } 177 177 + 178 178 + #[derive(Debug, Clone, Default)] 179 179 + /// 180 180 + /// OHLC packet structure 181 181 + /// 182 182 + pub struct OHLC { 183 183 + pub open: f64, 184 184 + pub high: f64, 185 185 + pub low: f64, 186 186 + pub close: f64, 187 187 + } 188 188 + 189 189 + impl OHLC { 190 190 + fn from(value: &[u8], exchange: &Exchange) -> Option<Self> { 191 191 + if let Some(bs) = value.get(0..16) { 192 192 + Some(OHLC { 193 193 + open: price(&bs[0..=3], exchange).unwrap(), 194 194 + high: price(&bs[4..=7], exchange).unwrap(), 195 195 + low: price(&bs[8..=11], exchange).unwrap(), 196 196 + close: price(&bs[12..=15], exchange).unwrap(), 197 197 + }) 198 198 + } else { 199 199 + None 200 200 + } 201 201 + } 202 202 + } 203 203 + 204 204 + #[derive(Debug, Clone, Default)] 205 205 + /// 206 206 + /// Market depth packet structure 207 207 + /// 208 208 + pub struct Depth { 209 209 + pub buy: [DepthItem; 5], 210 210 + pub sell: [DepthItem; 5], 211 211 + } 212 212 + 213 213 + impl Depth { 214 214 + fn from(input: &[u8], exchange: &Exchange) -> Option<Self> { 215 215 + if let Some(bs) = input.get(0..120) { 216 216 + let parse_depth_item = |v: &[u8], start: usize| { 217 217 + v.get(start..start + 10) 218 218 + .and_then(|xs| DepthItem::from(xs, exchange)) 219 219 + .unwrap_or_default() 220 220 + }; 221 221 + let mut depth = Depth::default(); 222 222 + for i in 0..5 { 223 223 + let start = i * 12; 224 224 + depth.buy[i] = parse_depth_item(bs, start) 225 225 + } 226 226 + for i in 0..5 { 227 227 + let start = 60 + i * 12; 228 228 + depth.sell[i] = parse_depth_item(bs, start); 229 229 + } 230 230 + 231 231 + Some(depth) 232 232 + } else { 233 233 + None 234 234 + } 235 235 + } 236 236 + } 237 237 + 238 238 + #[derive(Debug, Clone, Default)] 239 239 + /// 240 240 + /// Structure for each market depth entry 241 241 + /// 242 242 + pub struct DepthItem { 243 243 + pub qty: u32, 244 244 + pub price: f64, 245 245 + pub orders: u16, 246 246 + } 247 247 + 248 248 + impl DepthItem { 249 249 + pub fn from(input: &[u8], exchange: &Exchange) -> Option<Self> { 250 250 + if let Some(bs) = input.get(0..10) { 251 251 + Some(DepthItem { 252 252 + qty: value(&bs[0..=3]).unwrap(), 253 253 + price: price(&bs[4..=7], exchange).unwrap(), 254 254 + orders: value_short(&bs[8..=9]).unwrap(), 255 255 + }) 256 256 + } else { 257 257 + None 258 258 + } 259 259 + } 260 260 + } 261 261 + 262 262 + #[derive(Debug, Clone)] 263 263 + /// 264 264 + /// Parsed message from websocket 265 265 + /// 266 266 + pub enum TickerMessage { 267 267 + /// Quote packets for subscribed tokens 268 268 + Tick(Vec<TickMessage>), 269 269 + /// Error response 270 270 + Error(String), 271 271 + /// Order postback 272 272 + Order(serde_json::Value), 273 273 + /// Messages and alerts from broker 274 274 + Message(serde_json::Value), 275 275 + /// Websocket closing frame 276 276 + ClosingMessage(serde_json::Value), 277 277 + } 278 278 + 279 279 + impl From<TextMessage> for TickerMessage { 280 280 + fn from(value: TextMessage) -> Self { 281 281 + let message_type: TextMessageType = value.message_type.into(); 282 282 + match message_type { 283 283 + TextMessageType::Order => Self::Order(value.data), 284 284 + TextMessageType::Error => Self::Error(value.data.to_string()), 285 285 + TextMessageType::Message => Self::Message(value.data), 286 286 + } 287 287 + } 288 288 + } 289 289 + 290 290 + #[derive(Debug, Clone, Default)] 291 291 + /// 292 292 + /// Parsed quote packet 293 293 + /// 294 294 + pub struct TickMessage { 295 295 + pub instrument_token: u32, 296 296 + pub content: Tick, 297 297 + } 298 298 + 299 299 + impl TickMessage { 300 300 + pub(crate) fn new(instrument_token: u32, content: Tick) -> Self { 301 301 + Self { 302 302 + instrument_token, 303 303 + content, 304 304 + } 305 305 + } 306 306 + } 307 307 + 308 308 + #[derive(Debug, Clone, Default)] 309 309 + /// 310 310 + /// Exchange options 311 311 + /// 312 312 + pub enum Exchange { 313 313 + #[default] 314 314 + NSE, 315 315 + NFO, 316 316 + CDS, 317 317 + BSE, 318 318 + BFO, 319 319 + BCD, 320 320 + MCX, 321 321 + MCXSX, 322 322 + INDICES, 323 323 + } 324 324 + 325 325 + impl Exchange { 326 326 + fn divisor(&self) -> f64 { 327 327 + match self { 328 328 + Self::CDS => 100_000_0.0, 329 329 + Self::BCD => 100_0.0, 330 330 + _ => 100.0, 331 331 + } 332 332 + } 333 333 + 334 334 + fn is_tradable(&self) -> bool { 335 335 + match self { 336 336 + Self::INDICES => false, 337 337 + _ => true, 338 338 + } 339 339 + } 340 340 + } 341 341 + 342 342 + impl From<usize> for Exchange { 343 343 + fn from(value: usize) -> Self { 344 344 + match value { 345 345 + 9 => Self::INDICES, 346 346 + 8 => Self::MCXSX, 347 347 + 7 => Self::MCX, 348 348 + 6 => Self::BCD, 349 349 + 5 => Self::BFO, 350 350 + 4 => Self::BSE, 351 351 + 3 => Self::CDS, 352 352 + 2 => Self::NFO, 353 353 + 1 => Self::NSE, 354 354 + _ => Self::NSE, 355 355 + } 356 356 + } 357 357 + } 358 358 + 359 359 + #[derive( 360 360 + Debug, Clone, Deserialize, Serialize, Default, PartialEq, PartialOrd, 361 361 + )] 362 362 + #[serde(rename_all = "lowercase")] 363 363 + /// 364 364 + /// Modes in which packets are streamed 365 365 + /// 366 366 + pub enum Mode { 367 367 + Full, 368 368 + #[default] 369 369 + Quote, 370 370 + LTP, 371 371 + } 372 372 + 373 373 + impl TryFrom<usize> for Mode { 374 374 + type Error = String; 375 375 + fn try_from(value: usize) -> Result<Self, Self::Error> { 376 376 + match value { 377 377 + 8 => Ok(Self::LTP), 378 378 + 44 => Ok(Self::Quote), 379 379 + 184 => Ok(Self::Full), 380 380 + _ => Err(format!("Invalid packet size: {}", value)), 381 381 + } 382 382 + } 383 383 + } 384 384 + 385 385 + #[derive(Clone, Debug, Deserialize, Serialize)] 386 386 + #[serde(rename_all = "lowercase")] 387 387 + /// 388 388 + /// Websocket request actions 389 389 + /// 390 390 + enum RequestActions { 391 391 + Subscribe, 392 392 + Unsubscribe, 393 393 + Mode, 394 394 + } 395 395 + 396 396 + #[derive(Clone, Debug, Deserialize, Serialize)] 397 397 + #[serde(untagged)] 398 398 + /// 399 399 + /// Websocket request data 400 400 + /// 401 401 + enum RequestData { 402 402 + InstrumentTokens(Vec<u32>), 403 403 + InstrumentTokensWithMode(Mode, Vec<u32>), 404 404 + } 405 405 + 406 406 + #[derive(Debug, Clone, Deserialize, Serialize)] 407 407 + /// 408 408 + /// Websocket request structure 409 409 + /// 410 410 + pub struct Request { 411 411 + a: RequestActions, 412 412 + v: RequestData, 413 413 + } 414 414 + 415 415 + impl Request { 416 416 + fn new(action: RequestActions, value: RequestData) -> Request { 417 417 + Request { 418 418 + a: action, 419 419 + v: value, 420 420 + } 421 421 + } 422 422 + 423 423 + /// 424 424 + /// Subscribe to a list of instrument tokens 425 425 + /// 426 426 + pub fn subscribe(instrument_tokens: Vec<u32>) -> Request { 427 427 + Request::new( 428 428 + RequestActions::Subscribe, 429 429 + RequestData::InstrumentTokens(instrument_tokens), 430 430 + ) 431 431 + } 432 432 + 433 433 + /// 434 434 + /// Subscribe to a list of instrument tokens with mode 435 435 + /// 436 436 + pub fn mode(mode: Mode, instrument_tokens: Vec<u32>) -> Request { 437 437 + Request::new( 438 438 + RequestActions::Mode, 439 439 + RequestData::InstrumentTokensWithMode(mode, instrument_tokens), 440 440 + ) 441 441 + } 442 442 + 443 443 + /// 444 444 + /// Unsubscribe from a list of instrument tokens 445 445 + /// 446 446 + pub fn unsubscribe(instrument_tokens: Vec<u32>) -> Request { 447 447 + Request::new( 448 448 + RequestActions::Unsubscribe, 449 449 + RequestData::InstrumentTokens(instrument_tokens), 450 450 + ) 451 451 + } 452 452 + } 453 453 + 454 454 + impl ToString for Request { 455 455 + fn to_string(&self) -> String { 456 456 + serde_json::to_string(self) 457 457 + .expect("failed to serialize TickerInput to JSON") 458 458 + } 459 459 + } 460 460 + 461 461 + #[derive(Debug, Clone)] 462 462 + /// 463 463 + /// Postbacks and non-binary message types 464 464 + /// 465 465 + enum TextMessageType { 466 466 + /// Order postback 467 467 + Order, 468 468 + /// Error response 469 469 + Error, 470 470 + /// Messages and alerts from the broker 471 471 + Message, 472 472 + } 473 473 + 474 474 + impl From<String> for TextMessageType { 475 475 + fn from(value: String) -> Self { 476 476 + match value.as_str() { 477 477 + "order" => Self::Order, 478 478 + "error" => Self::Error, 479 479 + _ => Self::Message, 480 480 + } 481 481 + } 482 482 + } 483 483 + 484 484 + #[derive(Debug, Clone, Deserialize, Serialize)] 485 485 + /// 486 486 + /// Postback and non-binary message structure 487 487 + /// 488 488 + pub struct TextMessage { 489 489 + #[serde(rename = "type")] 490 490 + message_type: String, 491 491 + data: serde_json::Value, 492 492 + }

+476

src/ticker.rs

reviewed

··· 1 1 + use futures_util::{stream::iter, SinkExt, StreamExt}; 2 2 + use serde_json::json; 3 3 + use std::{collections::HashMap, sync::Arc}; 4 4 + use tokio::net::TcpStream; 5 5 + use tokio::sync::Mutex; 6 6 + use tokio_tungstenite::{ 7 7 + connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream, 8 8 + }; 9 9 + 10 10 + use crate::models::{ 11 11 + packet_length, Mode, Request, TextMessage, Tick, TickMessage, TickerMessage, 12 12 + }; 13 13 + 14 14 + #[derive(Debug, Clone)] 15 15 + /// 16 16 + /// The WebSocket client for connecting to Kite Connect's streaming quotes service. 17 17 + /// 18 18 + pub struct KiteTickerAsync { 19 19 + #[allow(dead_code)] 20 20 + api_key: String, 21 21 + #[allow(dead_code)] 22 22 + access_token: String, 23 23 + ws_stream: Arc<Mutex<WebSocketStream<MaybeTlsStream<TcpStream>>>>, 24 24 + } 25 25 + 26 26 + impl KiteTickerAsync { 27 27 + /// Establish a connection with the Kite WebSocket server 28 28 + pub async fn connect( 29 29 + api_key: &str, 30 30 + access_token: &str, 31 31 + ) -> Result<Self, String> { 32 32 + let socket_url = format!( 33 33 + "wss://{}?api_key={}&access_token={}", 34 34 + "ws.kite.trade", api_key, access_token 35 35 + ); 36 36 + let url = url::Url::parse(socket_url.as_str()).unwrap(); 37 37 + 38 38 + let (ws_stream, _) = connect_async(url).await.map_err(|e| e.to_string())?; 39 39 + 40 40 + Ok(KiteTickerAsync { 41 41 + api_key: api_key.to_string(), 42 42 + access_token: access_token.to_string(), 43 43 + ws_stream: Arc::new(Mutex::new(ws_stream)), 44 44 + }) 45 45 + } 46 46 + 47 47 + /// Subscribes the client to a list of instruments 48 48 + pub async fn subscribe( 49 49 + mut self, 50 50 + instrument_tokens: &[u32], 51 51 + mode: Option<Mode>, 52 52 + ) -> Result<KiteTickerSubscriber, String> { 53 53 + self 54 54 + .subscribe_cmd(instrument_tokens, mode.clone()) 55 55 + .await 56 56 + .expect("failed to subscribe"); 57 57 + let st = instrument_tokens 58 58 + .to_vec() 59 59 + .iter() 60 60 + .map(|t| (t.clone(), mode.to_owned().unwrap_or_default())) 61 61 + .collect(); 62 62 + 63 63 + Ok(KiteTickerSubscriber { 64 64 + ticker: self, 65 65 + subscribed_tokens: st, 66 66 + }) 67 67 + } 68 68 + 69 69 + /// Close the websocket connection 70 70 + pub async fn close(&mut self) -> Result<(), String> { 71 71 + let mut ws_stream = self.ws_stream.lock().await; 72 72 + ws_stream.close(None).await.map_err(|x| x.to_string())?; 73 73 + Ok(()) 74 74 + } 75 75 + 76 76 + async fn subscribe_cmd( 77 77 + &mut self, 78 78 + instrument_tokens: &[u32], 79 79 + mode: Option<Mode>, 80 80 + ) -> Result<(), String> { 81 81 + let mut msgs = iter(vec![ 82 82 + Ok(Message::Text( 83 83 + Request::subscribe(instrument_tokens.to_vec()).to_string(), 84 84 + )), 85 85 + Ok(Message::Text( 86 86 + Request::mode(mode.unwrap_or_default(), instrument_tokens.to_vec()) 87 87 + .to_string(), 88 88 + )), 89 89 + ]); 90 90 + 91 91 + let mut ws_stream = self.ws_stream.lock().await; 92 92 + 93 93 + ws_stream 94 94 + .send_all(msgs.by_ref()) 95 95 + .await 96 96 + .expect("failed to send subscription message"); 97 97 + 98 98 + Ok(()) 99 99 + } 100 100 + 101 101 + async fn unsubscribe_cmd( 102 102 + &mut self, 103 103 + instrument_tokens: &[u32], 104 104 + ) -> Result<(), String> { 105 105 + let mut ws_stream = self.ws_stream.lock().await; 106 106 + ws_stream 107 107 + .send(Message::Text( 108 108 + Request::unsubscribe(instrument_tokens.to_vec()).to_string(), 109 109 + )) 110 110 + .await 111 111 + .expect("failed to send unsubscribe message"); 112 112 + Ok(()) 113 113 + } 114 114 + 115 115 + async fn set_mode_cmd( 116 116 + &mut self, 117 117 + instrument_tokens: &[u32], 118 118 + mode: Mode, 119 119 + ) -> Result<(), String> { 120 120 + let mut ws_stream = self.ws_stream.lock().await; 121 121 + ws_stream 122 122 + .send(Message::Text( 123 123 + Request::mode(mode, instrument_tokens.to_vec()).to_string(), 124 124 + )) 125 125 + .await 126 126 + .expect("failed to send set mode message"); 127 127 + Ok(()) 128 128 + } 129 129 + } 130 130 + 131 131 + #[derive(Debug, Clone)] 132 132 + /// 133 133 + /// The Websocket client that entered in a pub/sub mode once the client subscribed to a list of instruments 134 134 + /// 135 135 + pub struct KiteTickerSubscriber { 136 136 + ticker: KiteTickerAsync, 137 137 + subscribed_tokens: HashMap<u32, Mode>, 138 138 + } 139 139 + 140 140 + impl KiteTickerSubscriber { 141 141 + /// Get the list of subscribed instruments 142 142 + pub fn get_subscribed(&self) -> Vec<u32> { 143 143 + self 144 144 + .subscribed_tokens 145 145 + .clone() 146 146 + .into_keys() 147 147 + .collect::<Vec<_>>() 148 148 + } 149 149 + 150 150 + /// get all tokens common between subscribed tokens and input tokens 151 151 + /// and if the input is empty then all subscribed tokens will be unsubscribed 152 152 + fn get_subscribed_or(&self, tokens: &[u32]) -> Vec<u32> { 153 153 + if tokens.len() == 0 { 154 154 + self.get_subscribed() 155 155 + } else { 156 156 + tokens 157 157 + .iter() 158 158 + .filter(|t| self.subscribed_tokens.contains_key(t)) 159 159 + .map(|t| t.clone()) 160 160 + .collect::<Vec<_>>() 161 161 + } 162 162 + } 163 163 + 164 164 + /// Subscribe to new tokens 165 165 + pub async fn subscribe( 166 166 + &mut self, 167 167 + tokens: &[u32], 168 168 + mode: Option<Mode>, 169 169 + ) -> Result<(), String> { 170 170 + self.subscribed_tokens.extend( 171 171 + tokens 172 172 + .iter() 173 173 + .map(|t| (t.clone(), mode.clone().unwrap_or_default())), 174 174 + ); 175 175 + let tks = self.get_subscribed(); 176 176 + dbg!(tks.clone()); 177 177 + self.ticker.subscribe_cmd(tks.as_slice(), None).await?; 178 178 + Ok(()) 179 179 + } 180 180 + 181 181 + /// Change the mode of the subscribed instrument tokens 182 182 + pub async fn set_mode( 183 183 + &mut self, 184 184 + instrument_tokens: &[u32], 185 185 + mode: Mode, 186 186 + ) -> Result<(), String> { 187 187 + let tokens = self.get_subscribed_or(instrument_tokens); 188 188 + self.ticker.set_mode_cmd(tokens.as_slice(), mode).await 189 189 + } 190 190 + 191 191 + /// Unsubscribe provided subscribed tokens, if input is empty then all subscribed tokens will unsubscribed 192 192 + /// 193 193 + /// Tokens in the input which are not part of the subscribed tokens will be ignored. 194 194 + pub async fn unsubscribe( 195 195 + &mut self, 196 196 + instrument_tokens: &[u32], 197 197 + ) -> Result<(), String> { 198 198 + let tokens = self.get_subscribed_or(instrument_tokens); 199 199 + dbg!(tokens.clone()); 200 200 + self.ticker.unsubscribe_cmd(tokens.as_slice()).await 201 201 + } 202 202 + 203 203 + /// Get the next message from the server, waiting if necessary. 204 204 + /// If the result is None then server is terminated 205 205 + pub async fn next_message(&mut self) -> Result<Option<TickerMessage>, String> { 206 206 + let mut ws_stream = self.ticker.ws_stream.lock().await; 207 207 + match ws_stream.next().await { 208 208 + Some(message) => match message { 209 209 + Ok(msg) => Ok(self.process_message(msg)), 210 210 + Err(e) => Err(e.to_string()), 211 211 + }, 212 212 + None => Ok(None), 213 213 + } 214 214 + } 215 215 + 216 216 + fn process_message(&self, message: Message) -> Option<TickerMessage> { 217 217 + match message { 218 218 + Message::Text(text_message) => self.process_text_message(text_message), 219 219 + Message::Binary(ref binary_message) => { 220 220 + if binary_message.len() < 2 { 221 221 + return Some(TickerMessage::Tick(vec![])); 222 222 + } else { 223 223 + self.process_binary(binary_message.as_slice()) 224 224 + } 225 225 + } 226 226 + Message::Close(closing_message) => closing_message.map(|c| { 227 227 + TickerMessage::ClosingMessage(json!({ 228 228 + "code": c.code.to_string(), 229 229 + "reason": c.reason.to_string() 230 230 + })) 231 231 + }), 232 232 + Message::Ping(_) => unimplemented!(), 233 233 + Message::Pong(_) => unimplemented!(), 234 234 + Message::Frame(_) => unimplemented!(), 235 235 + } 236 236 + } 237 237 + 238 238 + fn process_binary(&self, binary_message: &[u8]) -> Option<TickerMessage> { 239 239 + // 0 - 2 : number of packets in the message 240 240 + let num_packets = 241 241 + i16::from_be_bytes(binary_message[0..=1].try_into().unwrap()) as usize; 242 242 + if num_packets > 0 { 243 243 + Some(TickerMessage::Tick( 244 244 + [0..num_packets] 245 245 + .into_iter() 246 246 + .fold((vec![], 2), |(mut acc, start), _| { 247 247 + // start - start + 2 : length of the packet 248 248 + let packet_len = packet_length(&binary_message[start..start + 2]); 249 249 + let next_start = start + 2 + packet_len; 250 250 + let tick = Tick::from(&binary_message[start + 2..next_start]); 251 251 + acc.push(TickMessage::new(tick.instrument_token, tick)); 252 252 + (acc, next_start) 253 253 + }) 254 254 + .0, 255 255 + )) 256 256 + } else { 257 257 + None 258 258 + } 259 259 + } 260 260 + 261 261 + fn process_text_message( 262 262 + &self, 263 263 + text_message: String, 264 264 + ) -> Option<TickerMessage> { 265 265 + dbg!(text_message.clone()); 266 266 + serde_json::from_str::<TextMessage>(&text_message) 267 267 + .map(|x| x.into()) 268 268 + .ok() 269 269 + } 270 270 + } 271 271 + 272 272 + #[cfg(test)] 273 273 + mod tests { 274 274 + use super::*; 275 275 + use tokio::select; 276 276 + 277 277 + async fn check<F>( 278 278 + mode: Mode, 279 279 + token: u32, 280 280 + sb: &mut KiteTickerSubscriber, 281 281 + assertions: Option<F>, 282 282 + ) where 283 283 + F: Fn(Vec<TickMessage>) -> (), 284 284 + { 285 285 + loop { 286 286 + match sb.next_message().await { 287 287 + Ok(message) => match message { 288 288 + Some(TickerMessage::Tick(xs)) => { 289 289 + if xs.len() == 0 { 290 290 + continue; 291 291 + } 292 292 + assertions.map(|f| f(xs.clone())).or_else(|| { 293 293 + let tick_message = xs.first().unwrap(); 294 294 + assert!(tick_message.instrument_token == token); 295 295 + assert_eq!(tick_message.content.mode, mode); 296 296 + Some(()) 297 297 + }); 298 298 + break; 299 299 + } 300 300 + _ => { 301 301 + dbg!(message); 302 302 + continue; 303 303 + } 304 304 + }, 305 305 + _ => { 306 306 + assert!(false); 307 307 + break; 308 308 + } 309 309 + } 310 310 + } 311 311 + } 312 312 + 313 313 + #[tokio::test] 314 314 + async fn test_ticker() { 315 315 + let api_key = std::env::var("KITE_API_KEY").unwrap(); 316 316 + let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap(); 317 317 + let ticker = KiteTickerAsync::connect(&api_key, &access_token).await; 318 318 + 319 319 + assert_eq!(ticker.is_ok(), true); 320 320 + 321 321 + let ticker = ticker.unwrap(); 322 322 + let token = 94977; // bata 323 323 + let mode = Mode::Full; 324 324 + let sb = ticker.subscribe(&[token], Some(mode.clone())).await; 325 325 + assert_eq!(sb.is_ok(), true); 326 326 + let mut sb = sb.unwrap(); 327 327 + assert_eq!(sb.subscribed_tokens.len(), 1); 328 328 + let mut loop_cnt = 0; 329 329 + loop { 330 330 + loop_cnt += 1; 331 331 + select! { 332 332 + Ok(n) = sb.next_message() => { 333 333 + match n.to_owned() { 334 334 + Some(message) => { 335 335 + match message { 336 336 + TickerMessage::Tick(xs) => { 337 337 + if xs.len() == 0 { 338 338 + if loop_cnt > 5 { 339 339 + break; 340 340 + }else { 341 341 + continue; 342 342 + } 343 343 + } 344 344 + assert_eq!(xs.len(), 1); 345 345 + let tick_message = xs.first().unwrap(); 346 346 + dbg!(tick_message); 347 347 + assert!(tick_message.instrument_token == token); 348 348 + assert_eq!(tick_message.content.mode, mode); 349 349 + if loop_cnt > 5 { 350 350 + break; 351 351 + } 352 352 + }, 353 353 + _ => { 354 354 + if loop_cnt > 5 { 355 355 + break; 356 356 + } 357 357 + } 358 358 + } 359 359 + }, 360 360 + _ => { 361 361 + if loop_cnt > 5 { 362 362 + assert!(false); 363 363 + break; 364 364 + } 365 365 + } 366 366 + } 367 367 + }, 368 368 + else => { 369 369 + assert!(false); 370 370 + break; 371 371 + } 372 372 + } 373 373 + } 374 374 + 375 375 + sb.ticker.close().await.unwrap(); 376 376 + } 377 377 + 378 378 + #[tokio::test] 379 379 + async fn test_unsubscribe() { 380 380 + // create a ticker 381 381 + let api_key = std::env::var("KITE_API_KEY").unwrap(); 382 382 + let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap(); 383 383 + let ticker = KiteTickerAsync::connect(&api_key, &access_token).await; 384 384 + 385 385 + let ticker = ticker.unwrap(); 386 386 + let token = 94977; // bata 387 387 + let mode = Mode::Full; 388 388 + let mut sb = ticker 389 389 + .subscribe(&[token], Some(mode.clone())) 390 390 + .await 391 391 + .unwrap(); 392 392 + 393 393 + let mut loop_cnt = 0; 394 394 + 395 395 + loop { 396 396 + match sb.next_message().await { 397 397 + Ok(message) => match message { 398 398 + Some(TickerMessage::Tick(xs)) => { 399 399 + if xs.len() == 0 { 400 400 + if loop_cnt > 4 { 401 401 + assert!(true); 402 402 + break; 403 403 + } else { 404 404 + loop_cnt += 1; 405 405 + continue; 406 406 + } 407 407 + } 408 408 + assert_eq!(xs.len(), 1); 409 409 + let tick_message = xs.first().unwrap(); 410 410 + dbg!(tick_message); 411 411 + assert!(tick_message.instrument_token == token); 412 412 + sb.unsubscribe(&[]).await.unwrap(); 413 413 + loop_cnt += 1; 414 414 + if loop_cnt > 5 { 415 415 + assert!(false); 416 416 + break; 417 417 + } 418 418 + } 419 419 + _ => { 420 420 + dbg!(message); 421 421 + continue; 422 422 + } 423 423 + }, 424 424 + _ => { 425 425 + assert!(false); 426 426 + break; 427 427 + } 428 428 + } 429 429 + } 430 430 + sb.ticker.close().await.unwrap(); 431 431 + } 432 432 + 433 433 + async fn create_ticker() -> KiteTickerAsync { 434 434 + // create a ticker 435 435 + let api_key = std::env::var("KITE_API_KEY").unwrap(); 436 436 + let access_token = std::env::var("KITE_ACCESS_TOKEN").unwrap(); 437 437 + let ticker = KiteTickerAsync::connect(&api_key, &access_token).await; 438 438 + ticker.expect("failed to create ticker") 439 439 + } 440 440 + 441 441 + #[tokio::test] 442 442 + async fn test_set_mode() { 443 443 + let ticker = create_ticker().await; 444 444 + let token = 94977; // bata 445 445 + let mode = Mode::LTP; 446 446 + let new_mode = Mode::Quote; 447 447 + let mut sb = ticker 448 448 + .subscribe(&[token], Some(mode.clone())) 449 449 + .await 450 450 + .unwrap(); 451 451 + 452 452 + let f1: Option<Box<dyn Fn(Vec<TickMessage>) -> ()>> = None; 453 453 + let f2: Option<Box<dyn Fn(Vec<TickMessage>) -> ()>> = None; 454 454 + check(mode, token, &mut sb, f1).await; 455 455 + sb.set_mode(&[], new_mode.clone()).await.unwrap(); 456 456 + check(new_mode, token, &mut sb, f2).await; 457 457 + 458 458 + sb.ticker.close().await.unwrap(); 459 459 + } 460 460 + 461 461 + #[tokio::test] 462 462 + async fn test_new_sub() { 463 463 + let ticker = create_ticker().await; 464 464 + let token = 94977; // bata 465 465 + let mode = Mode::LTP; 466 466 + let mut sb = ticker 467 467 + .subscribe(&[token], Some(mode.clone())) 468 468 + .await 469 469 + .unwrap(); 470 470 + tokio::spawn(async move { 471 471 + sb.subscribe(&[2953217], None).await.unwrap(); 472 472 + }) 473 473 + .await 474 474 + .unwrap(); 475 475 + } 476 476 + }