Merge pull request #2 from kolybelkin/feature/mbo

MBO familiar implementation of MDP3

.travis.yml

@@ -0,0 +1,3 @@
+language: java
+jdk:
+  - oraclejdk8

build.gradle

@@ -1,42 +1 @@
-plugins {
-    id "java"
-    id "maven"
-    id "java-library-distribution"
-    id "com.github.samueltbrown.cucumber" version "0.9"
-    id "org.sonarqube" version "2.0.1"
-    id "me.champeau.gradle.jmh" version "0.3.0"
-}
-group = 'com.epam.cme'
-version = '1.0'
-sourceCompatibility = 1.8
-targetCompatibility = 1.8
-
-repositories {
-    mavenCentral()
-}
-
-
-dependencies {
-    compile group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.5'
-    compile group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.5'
-    runtime group: 'org.apache.logging.log4j', name: 'log4j-slf4j-impl', version: '2.5'
-    compile group: 'commons-configuration', name: 'commons-configuration', version: '1.10'
-    runtime group: 'commons-collections', name: 'commons-collections', version: '3.2.2'
-    compile group: 'net.openhft', name: 'chronicle-bytes', version: '1.7.17'
-    compile group: 'net.openhft', name: 'koloboke-api-jdk8', version: '0.6.8'
-    runtime group: 'net.openhft', name: 'koloboke-impl-jdk8', version: '0.6.8'
-    compile group: 'net.openhft', name: 'chronicle-core', version: '1.7.6'
-
-    cucumberCompile 'info.cukes:cucumber-java:1.2.4'
-    cucumberCompile 'info.cukes:cucumber-picocontainer:1.2.4'
-    testCompile group: 'junit', name: 'junit', version: '4.11'
-}
-
-distributions {
-    main {
-        baseName = 'b2bits-jmdp3'
-        contents {
-            from { 'dist' }
-        }
-    }
-}
+defaultTasks 'clean', 'build', 'distZip', 'distTar'

core/build.gradle

@@ -0,0 +1,21 @@
+group 'com.epam.cme'
+version '2.0'
+
+apply plugin: 'java'
+
+sourceCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+jar {
+    baseName 'b2bits-jmdp3-core'
+}
+
+dependencies {
+    compile group: 'net.openhft', name: 'chronicle-bytes', version: '1.7.17'
+    compile group: 'commons-configuration', name: 'commons-configuration', version: '1.10'
+    runtime group: 'commons-collections', name: 'commons-collections', version: '3.2.2'
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+}

core/src/main/java/com/epam/cme/mdp3/MdConstants.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2004-2016 EPAM Systems
+ * This file is part of Java Market Data Handler for CME Market Data (MDP 3.0).
+ * Java Market Data Handler for CME Market Data (MDP 3.0) is free software: you can redistribute it and/or modify it under the terms of the
+ * GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
+ * Java Market Data Handler for CME Market Data (MDP 3.0) is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License along with Java Market Data Handler for CME Market Data (MDP 3.0).
+ * If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.epam.cme.mdp3;
+
+public interface MdConstants {
+    byte TOP_OF_THE_BOOK_LEVEL = 1;
+
+    int LAST_MSG_SEQ_NUM_PROCESSED = 369;
+    int SECURITY_ID = 48;
+    int SEC_DESC_TAG = 55;
+    int RPT_SEQ_NUM = 83;
+    int NO_MD_ENTRIES = 268;
+    int INCR_RFRSH_MD_ENTRY_TYPE = 269;
+    int INCR_RFRSH_MD_ACTION = 279;
+    int INCR_RFRSH_MD_PRICE_LEVEL = 1023;
+    int INCR_RFRSH_MD_ENTRY_PX = 270;
+    int INCR_RFRSH_MD_ENTRY_SIZE = 271;
+    int INCR_RFRSH_MD_ORDER_NUM = 346;
+    int NO_ORDER_ID_ENTRIES = 37705;
+    int NO_CHUNKS = 37709;
+    int CURRENT_CHUNK = 37710;
+    int TOT_NUM_REPORTS = 911;
+    int REFERENCE_ID = 9633;
+}

src/main/java/com/epam/cme/mdp3/MdpGroup.java → core/src/main/java/com/epam/cme/mdp3/MdpGroup.java

@@ -49,6 +49,14 @@ public interface MdpGroup extends FieldSet {
      */
     void getEntry(MdpGroupEntry groupEntry);
 
+    /**
+     * Copies data of defined MDP Group entry to the given instance.
+     *
+     * @param entryNum number of entry
+     * @param groupEntry MDP Group Entry
+     */
+    void getEntry(int entryNum, MdpGroupEntry groupEntry);
+
     /**
      * Creates copy of the current Mdp Group
      *

src/main/java/com/epam/cme/mdp3/MdpPacket.java → core/src/main/java/com/epam/cme/mdp3/MdpPacket.java

@@ -57,8 +57,12 @@ public static MdpPacket instance() {
      * @return new instance of MDP Packet
      */
     public static MdpPacket allocate() {
+        return allocate(MDP_PACKET_MAX_SIZE);
+    }
+
+    public static MdpPacket allocate(int size) {
         final MdpPacket packet = instance();
-        final ByteBuffer byteBuffer = ByteBuffer.allocateDirect(SbeConstants.MDP_PACKET_MAX_SIZE).order(ByteOrder.LITTLE_ENDIAN);
+        final ByteBuffer byteBuffer = ByteBuffer.allocateDirect(size).order(ByteOrder.LITTLE_ENDIAN);
         packet.wrapFromBuffer(byteBuffer);
         return packet;
     }
@@ -69,7 +73,7 @@ public static MdpPacket allocate() {
      * @return copy instance of MDP Packet
      */
     public MdpPacket copy() {
-        final MdpPacket copyInstance = allocate();
+        final MdpPacket copyInstance = allocate(buffer().length());
         copyInstance.buffer().copyFrom(this.buffer());
         return copyInstance;
     }

core/src/main/java/com/epam/cme/mdp3/core/channel/CoreChannelListener.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2004-2016 EPAM Systems
+ * This file is part of Java Market Data Handler for CME Market Data (MDP 3.0).
+ * Java Market Data Handler for CME Market Data (MDP 3.0) is free software: you can redistribute it and/or modify it under the terms of the
+ * GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
+ * Java Market Data Handler for CME Market Data (MDP 3.0) is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License along with Java Market Data Handler for CME Market Data (MDP 3.0).
+ * If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.epam.cme.mdp3.core.channel;
+
+
+import com.epam.cme.mdp3.*;
+
+public interface CoreChannelListener {
+    /**
+     * Called when a Channel Feed is started.
+     *
+     * @param channelId ID of MDP Channel
+     * @param feedType  Type of MDP Feed (e.g. snapshot or incremental)
+     * @param feed      Feed (e.g. A or B)
+     */
+    void onFeedStarted(final String channelId, final FeedType feedType, final Feed feed);
+
+    /**
+     * Called when a Channel Feed is stopped.
+     *
+     * @param channelId ID of MDP Channel
+     * @param feedType  type of MDP Feed (e.g. snapshot or incremental)
+     * @param feed      feed (e.g. A or B)
+     */
+    void onFeedStopped(final String channelId, final FeedType feedType, final Feed feed);
+
+    /**
+     * Called when a Channel Feed received new MDP Packet.
+     *
+     * @param channelId ID of MDP Channel
+     * @param feedType  Type of MDP Feed (e.g. snapshot or incremental)
+     * @param feed      Feed (e.g. A or B)
+     * @param mdpPacket MDP Packet which is just received and will be handled after this callback
+     */
+    void onPacket(final String channelId, final FeedType feedType, final Feed feed, final MdpPacket mdpPacket);
+
+    /**
+     * Called when a Channel has to start its reset.
+     *
+     * @param channelId    ID of MDP Channel
+     * @param resetMessage MDP Reset Message
+     */
+    void onBeforeChannelReset(final String channelId, final MdpMessage resetMessage);
+
+    /**
+     * Called when a Channel just finished its reset.
+     *
+     * @param channelId    ID of MDP Channel
+     * @param resetMessage MDP Reset Message
+     */
+    void onFinishedChannelReset(final String channelId, final MdpMessage resetMessage);
+
+    /**
+     * Called when a Channel state is changed.
+     *
+     * @param channelId ID of MDP Channel
+     * @param prevState Previous state
+     * @param newState  New state
+     */
+    void onChannelStateChanged(final String channelId, final ChannelState prevState, final ChannelState newState);
+
+    /**
+     * Called when MDP Security Definition Message is received and processed.
+     *
+     * @param channelId     ID of MDP Channel
+     * @param secDefMessage MDP Security Definition Message
+     * @return Required Subscription flags
+     * @see MdEventFlags
+     */
+    int onSecurityDefinition(final String channelId, final MdpMessage secDefMessage);
+
+    /**
+     * Called when MDP RFQ Message is received and processed.
+     *
+     * @param channelId  ID of MDP Channel
+     * @param rfqMessage MDP TFQ Message
+     */
+    void onRequestForQuote(final String channelId, final MdpMessage rfqMessage);
+
+    /**
+     * Called when MDP Security Status Message is received and processed.
+     *
+     * @param channelId        ID of MDP Channel
+     * @param securityId       Security ID
+     * @param secStatusMessage MDP Security Status Message
+     */
+    void onSecurityStatus(final String channelId, final int securityId, final MdpMessage secStatusMessage);
+}

src/main/java/com/epam/cme/mdp3/core/channel/MdpFeedContext.java → core/src/main/java/com/epam/cme/mdp3/core/channel/MdpFeedContext.java

@@ -13,6 +13,7 @@
 package com.epam.cme.mdp3.core.channel;
 
 import com.epam.cme.mdp3.*;
+import com.epam.cme.mdp3.core.cfg.ConnectionCfg;
 import com.epam.cme.mdp3.sbe.message.SbeDouble;
 import com.epam.cme.mdp3.sbe.message.SbeGroup;
 import com.epam.cme.mdp3.sbe.message.SbeGroupEntry;
@@ -32,9 +33,9 @@ public MdpFeedContext(final Feed feed, final FeedType feedType) {
         this.feedType = feedType;
     }
 
-    public MdpFeedContext(final MdpFeedWorker feedConnection) {
-        this.feed = feedConnection.getCfg().getFeed();
-        this.feedType = feedConnection.getCfg().getFeedType();
+    public MdpFeedContext(final ConnectionCfg cfg) {
+        this.feed = cfg.getFeed();
+        this.feedType = cfg.getFeedType();
     }
 
     public MdpPacket getMdpPacket() {

core/src/main/java/com/epam/cme/mdp3/core/channel/MdpFeedRtmState.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright 2004-2016 EPAM Systems
+ * This file is part of Java Market Data Handler for CME Market Data (MDP 3.0).
+ * Java Market Data Handler for CME Market Data (MDP 3.0) is free software: you can redistribute it and/or modify it under the terms of the
+ * GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
+ * Java Market Data Handler for CME Market Data (MDP 3.0) is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License along with Java Market Data Handler for CME Market Data (MDP 3.0).
+ * If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.epam.cme.mdp3.core.channel;
+
+public enum MdpFeedRtmState {
+    ACTIVE, PENDING_SHUTDOWN, SHUTDOWN, STOPPED
+}

src/main/java/com/epam/cme/mdp3/core/channel/MdpFeedWorker.java → core/src/main/java/com/epam/cme/mdp3/core/channel/MdpFeedWorker.java

@@ -66,7 +66,7 @@ public MdpFeedWorker(final ConnectionCfg cfg, final String networkInterface, fin
 
     void init() throws MdpFeedException {
         try {
-            this.feedContext = new MdpFeedContext(this);
+            this.feedContext = new MdpFeedContext(getCfg());
             this.ni = this.networkInterface != null ? NetworkInterface.getByName(this.networkInterface) : NetworkInterface.getByInetAddress(InetAddress.getLocalHost());
         } catch (IOException e) {
             logger.error("Failed open DatagramChannel", e);

core/src/main/java/com/epam/cme/mdp3/core/channel/tcp/MdpTCPChannel.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2004-2016 EPAM Systems
+ * This file is part of Java Market Data Handler for CME Market Data (MDP 3.0).
+ * Java Market Data Handler for CME Market Data (MDP 3.0) is free software: you can redistribute it and/or modify it under the terms of the
+ * GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
+ * Java Market Data Handler for CME Market Data (MDP 3.0) is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License along with Java Market Data Handler for CME Market Data (MDP 3.0).
+ * If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.epam.cme.mdp3.core.channel.tcp;
+
+import com.epam.cme.mdp3.core.cfg.ConnectionCfg;
+import com.epam.cme.mdp3.core.channel.MdpFeedContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.List;
+
+public class MdpTCPChannel implements TCPChannel {
+    private static final Logger logger = LoggerFactory.getLogger(MdpTCPChannel.class);
+    private final ConnectionCfg cfg;
+    private volatile SocketChannel socketChannel;
+    private MdpFeedContext feedContext;
+
+    public MdpTCPChannel(final ConnectionCfg cfg) {
+        this.cfg = cfg;
+        this.feedContext = new MdpFeedContext(cfg);
+    }
+
+    @Override
+    public boolean connect() {
+        List<String> hostIPs = cfg.getHostIPs();
+        for (String hostIP : hostIPs) {
+            try {
+                final InetSocketAddress inetSocketAddress = new InetSocketAddress(hostIP, cfg.getPort());
+                socketChannel = SocketChannel.open(inetSocketAddress);
+                logger.debug("Connected to {}:{}", hostIP, cfg.getPort());
+                return true;
+            } catch (Exception e) {
+                logger.error("Failed to connect to {}:{}. Exception: {}", hostIP, cfg.getPort(), e.getMessage(), e);
+            }
+        }
+        return false;
+    }
+
+    @Override
+    public void disconnect() {
+        try {
+            socketChannel.close();
+        } catch (IOException e) {
+            logger.error(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public void send(ByteBuffer bb) throws IOException {
+        socketChannel.write(bb);
+    }
+
+    @Override
+    public int receive(ByteBuffer bb) throws IOException {
+        int readBytes = socketChannel.read(bb);
+        if(readBytes < 0) {
+            throw new EOFException("Length of last received bytes is less than zero '" + readBytes + "'");
+        } else {
+            return readBytes;
+        }
+    }
+
+    @Override
+    public MdpFeedContext getFeedContext() {
+        return feedContext;
+    }
+}