public class TagValueMessageSpliterator implements Spliterator<ByteBuffer> {

  public static class TagValueFieldSpliterator implements Spliterator<ByteBuffer> {
    private static final byte FIELD_DELIMITER = 0x01;
    private static final byte TAG_DELIMITER = '=';

    public static int getTag(ByteBuffer buffer) {

      int delimiterPos = BufferUtil.find(buffer, 0, TAG_DELIMITER);
      if (delimiterPos > 0) {
        int length = delimiterPos - buffer.position();
        byte[] tagBuffer = new byte[length];
        buffer.get(tagBuffer, 0, length);
        //consumer the delimiter
        buffer.get();
        return Integer.parseInt(new String(tagBuffer));
      } else {
        return -1;
      }
    }

    public static byte[] getValue(ByteBuffer buffer) {
      int bytes = buffer.remaining();
      byte[] valueBuffer = new byte[bytes];
      buffer.get(valueBuffer, 0, bytes);
      return valueBuffer;
    }

    private final ByteBuffer buffer;
    private int offset;

    public TagValueFieldSpliterator(ByteBuffer buffer) {
      this.buffer = buffer;
      this.offset = 0;
    }

    @Override
    public int characteristics() {
      // TODO Auto-generated method stub
      return 0;
    }

    @Override
    public long estimateSize() {
      // TODO Auto-generated method stub
      return 0;
    }
    
    void reset() {
      offset = 0;
    }

    @Override
    public boolean tryAdvance(Consumer<? super ByteBuffer> consumer) {
      if (offset >= buffer.limit()) {
        return false;
      }
      int start = offset;
      int end = BufferUtil.find(buffer, this.offset, FIELD_DELIMITER);
      if (end > start) {
        try {
          ByteBuffer fieldBuffer = buffer.duplicate();
          fieldBuffer.position(start);
          fieldBuffer.limit(end);
          consumer.accept(fieldBuffer);
          this.offset = end + 1;
          return true;
        } catch (Exception ex) {
          ex.printStackTrace();
          return false;
        }
      } else {
        return false;
      }

    }

    @Override
    public Spliterator<ByteBuffer> trySplit() {
      // TODO Auto-generated method stub
      return null;
    }
  }


  private static final byte[] MESSAGE_DELIMITER = {0x0a};

  private final ByteBuffer buffer;
  private final TagValueFieldSpliterator fieldSpliter;
  private final SeekableByteChannel source;

  /**
   * Constructor for a file soucre
   * 
   * @param source byte channel
   * 
   *        todo: create a version that works with a supplied buffer
   */
  public TagValueMessageSpliterator(SeekableByteChannel source) {
    this.source = source;
    this.buffer = ByteBuffer.allocate(4096);
    this.fieldSpliter = new TagValueFieldSpliterator(buffer);
  }

  @Override
  public int characteristics() {
    // TODO Auto-generated method stub
    return 0;
  }

  @Override
  public long estimateSize() {
    // TODO Auto-generated method stub
    return 0;
  }

  @Override
  public boolean tryAdvance(Consumer<? super ByteBuffer> consumer) {
    try {
      long position = source.position();
      buffer.clear();
      int bytesRead = source.read(buffer);
      if (bytesRead > 0) {
        buffer.flip();
        int index = BufferUtil.find(buffer, 1, MESSAGE_DELIMITER);
        if (index < 0) {
          index = buffer.limit();
        }
        if (index == 0) {
          return false;
        }
        buffer.limit(index);
        consumer.accept(buffer);
        source.position(position + index + MESSAGE_DELIMITER.length);
        return true;
      }
      return false;
    } catch (IOException e) {
      return false;
    }
  }

  @Override
  public TagValueFieldSpliterator trySplit() {
    fieldSpliter.reset();
    return fieldSpliter;
  }

} public static class TagValueFieldSpliterator implements Spliterator<ByteBuffer> {
  private static final byte FIELD_DELIMITER = 0x01;
  private static final byte TAG_DELIMITER = '=';

  public static int getTag(ByteBuffer buffer) {

    int delimiterPos = BufferUtil.find(buffer, 0, TAG_DELIMITER);
    if (delimiterPos > 0) {
      int length = delimiterPos - buffer.position();
      byte[] tagBuffer = new byte[length];
      buffer.get(tagBuffer, 0, length);
      //consumer the delimiter
      buffer.get();
      return Integer.parseInt(new String(tagBuffer));
    } else {
      return -1;
    }
  }

  public static byte[] getValue(ByteBuffer buffer) {
    int bytes = buffer.remaining();
    byte[] valueBuffer = new byte[bytes];
    buffer.get(valueBuffer, 0, bytes);
    return valueBuffer;
  }

  private final ByteBuffer buffer;
  private int offset;

  public TagValueFieldSpliterator(ByteBuffer buffer) {
    this.buffer = buffer;
    this.offset = 0;
  }

  @Override
  public int characteristics() {
    // TODO Auto-generated method stub
    return 0;
  }

  @Override
  public long estimateSize() {
    // TODO Auto-generated method stub
    return 0;
  }
    
  void reset() {
    offset = 0;
  }

  @Override
  public boolean tryAdvance(Consumer<? super ByteBuffer> consumer) {
    if (offset >= buffer.limit()) {
      return false;
    }
    int start = offset;
    int end = BufferUtil.find(buffer, this.offset, FIELD_DELIMITER);
    if (end > start) {
      try {
        ByteBuffer fieldBuffer = buffer.duplicate();
        fieldBuffer.position(start);
        fieldBuffer.limit(end);
        consumer.accept(fieldBuffer);
        this.offset = end + 1;
        return true;
      } catch (Exception ex) {
        ex.printStackTrace();
        return false;
      }
    } else {
      return false;
    }

  }

  @Override
  public Spliterator<ByteBuffer> trySplit() {
    // TODO Auto-generated method stub
    return null;
  }
} public class MongoDBSecurityDefinitionStore implements SecurityDefinitionStore {

  class SecurityDefinitionBlock implements Block<Document> {

    private final Consumer<SecurityDefinitionImpl> consumer;
    private int count = 0;

    public SecurityDefinitionBlock(Consumer<SecurityDefinitionImpl> consumer) {
      this.consumer = consumer;
    }

    @Override
    public void apply(Document d) {
      count++;
      SecurityDefinitionImpl securityDefinition =
          jsonCodec.decode(d.toJson(), SecurityDefinitionImpl.class);
      consumer.accept(securityDefinition);
    }

    public int getCount() {
      return count;
    }
  }

  private static final String DATABASE_NAME = "refdata";
  private static final String SECDEF_COLLECTION_NAME = "secdef";
  private MongoCollection<Document> collection;
  private MongoDatabase database;
  private String[] hosts = {"localhost"};
  private final JsonCodec jsonCodec = new JsonCodec();
  private MongoClient mongoClient;

  /**
   * Constructor with default localhost
   */
  public MongoDBSecurityDefinitionStore() {}

  /**
   * Constructor with specified hosts
   * 
   * @param hosts array of host addresses of MongoDB cluster
   */
  public MongoDBSecurityDefinitionStore(String[] hosts) {
    this.hosts = hosts;
  }

  @Override
  public void close() {
    mongoClient.close();
  }

  public CompletableFuture<SecurityDefinitionStore> delete(
      SecurityDefinitionRequest securityDefinitionRequest) {

    CompletableFuture<SecurityDefinitionStore> future =
            new CompletableFuture<>();

    // todo: handle different request types
    collection.deleteMany(eq("symbol", securityDefinitionRequest.getSymbol()),
        (DeleteResult result, final Throwable t) -> {
          if (t == null) {
            future.complete(this);
          } else {
            future.completeExceptionally(t);
          }
        });
    return future;
  }

  @Override
  public CompletableFuture<SecurityDefinitionStore> insert(
      Collection<SecurityDefinition> securityDefinitions) {
    CompletableFuture<SecurityDefinitionStore> future =
            new CompletableFuture<>();

    List<Document> docs = securityDefinitions.stream().map(s -> {
      final String json = jsonCodec.encode(s);
      return Document.parse(json);
    }).collect(Collectors.toList());

    collection.insertMany(docs, (Void result, final Throwable t) -> {
      if (t == null) {
        future.complete(this);
      } else {
        future.completeExceptionally(t);
      }
    });
    return future;
  }

  @Override
  public CompletableFuture<SecurityDefinitionStore> insert(SecurityDefinition securityDefinition) {
    CompletableFuture<SecurityDefinitionStore> future =
            new CompletableFuture<>();

    final String json = jsonCodec.encode(securityDefinition);
    // convert json to bson
    Document doc = Document.parse(json);
    collection.insertOne(doc, (Void result, final Throwable t) -> {
      if (t == null) {
        future.complete(this);
      } else {
        future.completeExceptionally(t);
      }
    });
    return future;
  }

  @Override
  public CompletableFuture<SecurityDefinitionStore> open() {
    List<ServerAddress> hostList =
        Arrays.stream(hosts).map(h -> new ServerAddress(h)).collect(Collectors.toList());
    ClusterSettings clusterSettings = ClusterSettings.builder().hosts(hostList).build();
    MongoClientSettings settings =
        MongoClientSettings.builder().clusterSettings(clusterSettings).build();
    mongoClient = MongoClients.create(settings);

    database = mongoClient.getDatabase(DATABASE_NAME);
    collection = database.getCollection(SECDEF_COLLECTION_NAME);

    // In the case of MongoDB, open is synchronous because it doesn't
    // actually communicate with the server until a query is invoked.
    return CompletableFuture.completedFuture(this);
  }

  @Override
  public CompletableFuture<Integer> select(SecurityDefinitionRequest securityDefinitionRequest,
      Consumer<? extends SecurityDefinition> consumer) {

    CompletableFuture<Integer> future = new CompletableFuture<>();

    Bson filter = null;
    SecurityRequestType securityRequestType = securityDefinitionRequest.getSecurityRequestType();
    switch (securityRequestType) {
      case RequestSecurityIdentityAndSpecifications:
        break;
      case RequestListSecurities:
        break;
      case Symbol:
        filter = eq("symbol", securityDefinitionRequest.getSymbol());
        break;
      case SecurityTypeAndOrCFICode: {
        String securityType = securityDefinitionRequest.getSecurityType();
        String cfiCode = securityDefinitionRequest.getCfiCode();
        if (securityType != null && cfiCode != null) {
          filter = Filters.or(eq("securityType", securityDefinitionRequest.getSecurityType()),
              eq("cfiCode", securityDefinitionRequest.getCfiCode()));
        } else if (securityType != null) {
          filter = eq("securityType", securityDefinitionRequest.getSecurityType());
        } else if (cfiCode != null) {
          filter = eq("cfiCode", securityDefinitionRequest.getCfiCode());
        } else {
          throw new IllegalArgumentException("Missing security type or CFI code");
        }
      }
        break;
      case Product: {
        String product = securityDefinitionRequest.getProduct();
        String securityGroup = securityDefinitionRequest.getSecurityGroup();
        if (product != null) {
          filter = eq("product", product);
        } else if (securityGroup != null) {
          filter = eq("securityGroup", securityGroup);
        } else {
          throw new IllegalArgumentException("Missing product");
        }
      }
        break;
      case TradingSessionID:
        filter = eq("tradingSessionID", securityDefinitionRequest.getTradingSessionID());
        break;
      case AllSecurities:
        filter = exists("symbol");
        break;
      case MarketIDOrMarketID: {
        String marketId = securityDefinitionRequest.getMarketID();
        String marketSegmentID = securityDefinitionRequest.getMarketSegmentID();
        String securityExchange = securityDefinitionRequest.getSecurityExchange();
        if (marketId != null) {
          filter = elemMatch("marketSegmentGrp", Filters.eq("marketID", marketId));
        } else if (marketSegmentID != null) {
          filter = elemMatch("marketSegmentGrp", Filters.eq("marketSegmentID", marketSegmentID));
        } else if (securityExchange != null) {
          filter = eq("securityExchange", securityExchange);
        } else {
          throw new IllegalArgumentException("Missing market or market segment");
        }
      }
        break;
      default:
        throw new UnsupportedOperationException("Request type not supported");
    }

    @SuppressWarnings("unchecked")
    SecurityDefinitionBlock block =
        new SecurityDefinitionBlock((Consumer<SecurityDefinitionImpl>) consumer);

    SingleResultCallback<Void> callback = (result, t) -> {
      if (t == null) {
        future.complete(block.getCount());
      } else {
        future.completeExceptionally(t);
      }
    };


    collection.find(filter).forEach(block, callback);

    return future;
  }

  @Override
  public CompletableFuture<SecurityDefinitionStore> truncate() {
    CompletableFuture<SecurityDefinitionStore> future =
            new CompletableFuture<>();

    collection.drop((Void result, final Throwable t) -> {
      if (t == null) {
        collection = database.getCollection(SECDEF_COLLECTION_NAME);

        future.complete(this);
      } else {
        future.completeExceptionally(t);
      }
    });
    return future;
  }


} public interface SecurityDefinitionStore extends AutoCloseable {

  /**
   * Delete a security definition that matches the characteristics of the request
   * 
   * @param securityDefinitionRequest a request
   * @return future provides asynchronous result
   */
  CompletableFuture<SecurityDefinitionStore> delete(
      SecurityDefinitionRequest securityDefinitionRequest);

  /**
   * Insert a collection of security definitions into the store
   * 
   * @param securityDefinitions a collection of populated objects
   * @return future provides asynchronous result
   */
  CompletableFuture<SecurityDefinitionStore> insert(
      Collection<SecurityDefinition> securityDefinitions);

  /**
   * Insert a single security definition into the store
   * 
   * @param securityDefinition a populated object
   * @return future provides asynchronous result
   */
  CompletableFuture<SecurityDefinitionStore> insert(SecurityDefinition securityDefinition);

  /**
   * Open communications with the data store. It must be opened successfully before any other
   * operations may be executed.
   * 
   * @return future provides asynchronous result
   */
  CompletableFuture<SecurityDefinitionStore> open();

  /**
   * Retrieve security definitions from the store that fulfill the request
   * 
   * @param securityDefinitionRequest a populated request
   * @param consumer of selected security definitions
   * @return future signals when operation is complete and returns the total
   * number of security definitions returned
   */
  CompletableFuture<Integer> select(
      SecurityDefinitionRequest securityDefinitionRequest,
      Consumer<? extends SecurityDefinition> consumer);

  /**
   * Truncate <em>all</em> security definitions. Use with care.
   * 
   * @return future provides asynchronous result
   */
  CompletableFuture<SecurityDefinitionStore> truncate();
} public enum InstrAttribType {

  Callable(11), CallableBelowMaturityValue(21), CallableWithoutNotice(22), CouponPeriod(
      8), DummyInstrument(35), EscrowedToMaturity(12), EscrowedToRedemptionDate(13), Flat(
          1), InDefault(15), Indexed(18), InstrumentDenominator(25), InstrumentEligibleAnonOrders(
              30), InstrumentNumerator(26), InstrumentPricePrecision(27), InstrumentStrikePrice(
                  28), InterestBearing(3), LessFeeForPut(6), MinGuaranteedFillStatus(
                      32), MinGuaranteedFillVolume(31), NegativeSettlementPriceEligibility(
                          36), NegativeStrikePriceEligibility(37), NoPeriodicPayments(
                              4), OriginalIssueDiscount(10), OriginalIssueDiscountPrice(
                                  20), PreRefunded(14), PriceTickRulesForSecurity(
                                      23), SteppedCoupon(7), SubjectToAlternativeMinimumTax(
                                          19), Taxable(17), TestInstrument(34), Text(
                                              99), TradeableIndicator(
                                                  29), TradeAtSettlementEligibility(
                                                      33), TradeTypeEligibilityDetailsForSecurity(
                                                          24), Unrated(16), USStdContractInd(
                                                              38), VariableRate(
                                                                  5), When(9), ZeroCoupon(2);

  private static final EnumSet<InstrAttribType> valueSet = EnumSet.allOf(InstrAttribType.class);

  public static InstrAttribType getValue(int code) {
    Iterator<InstrAttribType> iter = valueSet.iterator();
    while (iter.hasNext()) {
      InstrAttribType value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final int code;

  InstrAttribType(int code) {
    this.code = code;
  }

  public int getCode() {
    return code;
  }

} public enum Product {
      AGENCY(1),
      COMMODITY(2),
      CORPORATE(3),
      CURRENCY(4),
      EQUITY(5),
      GOVERNMENT(6),
      INDEX(7),
      LOAN(8),
      MONEYMARKET(9),
      MORTGAGE(10),
      MUNICIPAL(11),
      OTHER(12),
      FINANCING(13);

  private static final EnumSet<Product> valueSet = EnumSet.allOf(Product.class);

  public static Product getValue(int code) {
    Iterator<Product> iter = valueSet.iterator();
    while (iter.hasNext()) {
      Product value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final int code;

  Product(int code) {
    this.code = code;
  }

  public int getCode() {
    return code;
  }
} public enum PutOrCall {
  Call(1), Put(0);

  private static final EnumSet<PutOrCall> valueSet = EnumSet.allOf(PutOrCall.class);

  public static PutOrCall getValue(int code) {
    Iterator<PutOrCall> iter = valueSet.iterator();
    while (iter.hasNext()) {
      PutOrCall value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final int code;

  PutOrCall(int code) {
    this.code = code;
  }

  public int getCode() {
    return code;
  }
} public class SecurityDefinitionImpl implements SecurityDefinition {

  public class InstrAttribImpl implements InstrAttrib {

    private InstrAttribType instrAttribType;
    private String instrAttribValue;

    @Override
    public InstrAttribType getInstrAttribType() {
      return instrAttribType;
    }

    @Override
    public String getInstrAttribValue() {
      return instrAttribValue;
    }

    public void setInstrAttribType(InstrAttribType instrAttribType) {
      this.instrAttribType = instrAttribType;
    }

    public void setInstrAttribValue(String instrAttribValue) {
      this.instrAttribValue = instrAttribValue;
    }
  }

  public class MarketSegmentImpl implements MarketSegment {

    private String marketID;
    private String marketSegmentID;


    @Override
    public String getMarketID() {
      return marketID;
    }

    @Override
    public String getMarketSegmentID() {
      return marketSegmentID;
    }


    public void setMarketID(String marketID) {
      this.marketID = marketID;
    }

    public void setMarketSegmentID(String marketSegmentID) {
      this.marketSegmentID = marketSegmentID;
    }
  }

  private final String msgType = "SecurityDefinition";

  private String cfiCode;
  private double contractMultiplier;
  private String currency;
  private final ArrayList<InstrAttribImpl> instrAttribGrp = new ArrayList<>();
  private LocalDateTime lastUpdateTime;
  private final ArrayList<MarketSegmentImpl> marketSegmentGrp = new ArrayList<>();
  private String matchAlgorithm;
  private String MaturityMonthYear;
  private int maxTradeVol;
  private double minPriceIncrement;
  private Product product;
  private PutOrCall putOrCall;
  private String SecurityDesc;
  private String securityExchange;
  private String securityGroup;
  private String SecurityID;
  private SecurityIDSource securityIDSource;
  private SecurityRejectReason securityRejectReason;
  private String securityReqID;
  private SecurityRequestResult securityRequestResult;
  private SecurityResponseType securityResponseType;
  private String securityType;
  private String settlCurrency;
  private SettlPriceType settlPriceType;
  private double strikePrice;
  private SubscriptionRequestType subscriptionRequestType;
  private String symbol;
  private String tradingSessionID;
  private String tradingSessionSubID;
  private Product underlyingProduct;

  public InstrAttribImpl addInstrAttrib() {
    InstrAttribImpl instrAttribImpl = new InstrAttribImpl();
    instrAttribGrp.add(instrAttribImpl);
    return instrAttribImpl;
  }

  public MarketSegmentImpl addMarketSegment() {
    MarketSegmentImpl marketSegmentImpl = new MarketSegmentImpl();
    marketSegmentGrp.add(marketSegmentImpl);
    return marketSegmentImpl;
  }

  @Override
  public String getCfiCode() {
    return cfiCode;
  }

  @Override
  public double getContractMultiplier() {
    return this.contractMultiplier;
  }

  @Override
  public String getCurrency() {
    return this.currency;
  }

  @Override
  public InstrAttrib getInstrAttrib(int index) {
    return instrAttribGrp.get(index);
  }

  @Override
  public LocalDateTime getLastUpdateTime() {
    return this.lastUpdateTime;
  }

  @Override
  public MarketSegment getMarketSegment(int index) {
    return marketSegmentGrp.get(index);
  }

  @Override
  public String getMatchAlgorithm() {
    return this.matchAlgorithm;
  }

  @Override
  public String getMaturityMonthYear() {
    return MaturityMonthYear;
  }

  @Override
  public int getMaxTradeVol() {
    return this.maxTradeVol;
  }

  @Override
  public double getMinPriceIncrement() {
    return this.minPriceIncrement;
  }

  @Override
  public String getMsgType() {
    return msgType;
  }

  @Override
  public int getNoInstrAttrib() {
    return instrAttribGrp.size();
  }

  @Override
  public int getNoMarketSegments() {
    return marketSegmentGrp.size();
  }

  @Override
  public Product getProduct() {
    return product;
  }

  @Override
  public PutOrCall getPutOrCall() {
    return putOrCall;
  }

  @Override
  public String getSecurityDesc() {
    return SecurityDesc;
  }

  @Override
  public String getSecurityExchange() {
    return this.securityExchange;
  }

  @Override
  public String getSecurityGroup() {
    return this.securityGroup;
  }

  @Override
  public String getSecurityID() {
    return SecurityID;
  }

  @Override
  public SecurityIDSource getSecurityIDSource() {
    return securityIDSource;
  }

  @Override
  public SecurityRejectReason getSecurityRejectReason() {
    return securityRejectReason;
  }

  @Override
  public String getSecurityReqID() {
    return securityReqID;
  }

  @Override
  public SecurityRequestResult getSecurityRequestResult() {
    return securityRequestResult;
  }

  @Override
  public SecurityResponseType getSecurityResponseType() {
    return securityResponseType;
  }

  @Override
  public String getSecurityType() {
    return securityType;
  }

  @Override
  public String getSettlCurrency() {
    return this.settlCurrency;
  }

  @Override
  public SettlPriceType getSettlPriceType() {
    return this.settlPriceType;
  }

  @Override
  public double getStrikePrice() {
    return strikePrice;
  }

  @Override
  public SubscriptionRequestType getSubscriptionRequestType() {
    return subscriptionRequestType;
  }

  @Override
  public String getSymbol() {
    return symbol;
  }

  @Override
  public String getTradingSessionID() {
    return tradingSessionID;
  }

  @Override
  public String getTradingSessionSubID() {
    return tradingSessionSubID;
  }

  @Override
  public Product getUnderlyingProduct() {
    return this.underlyingProduct;
  }

  public void setCfiCode(String cfiCode) {
    this.cfiCode = cfiCode;
  }

  public void setContractMultiplier(double contractMultiplier) {
    this.contractMultiplier = contractMultiplier;
  }

  public void setCurrency(String currency) {
    this.currency = currency;
  }

  public void setLastUpdateTime(LocalDateTime zonedDateTime) {
    this.lastUpdateTime = zonedDateTime;
  }

  public void setMatchAlgorithm(String matchAlgorithm) {
    this.matchAlgorithm = matchAlgorithm;
  }

  public void setMaturityMonthYear(String maturityMonthYear) {
    MaturityMonthYear = maturityMonthYear;
  }

  public void setMaxTradeVol(int maxTradeVol) {
    this.maxTradeVol = maxTradeVol;
  }

  public void setMinPriceIncrement(double minPriceIncrement) {
    this.minPriceIncrement = minPriceIncrement;
  }

  public void setProduct(Product product) {
    this.product = product;
  }

  public void setPutOrCall(PutOrCall putOrCall) {
    this.putOrCall = putOrCall;
  }

  public void setSecurityDesc(String securityDesc) {
    SecurityDesc = securityDesc;
  }

  public void setSecurityExchange(String securityExchange) {
    this.securityExchange = securityExchange;
  }

  public void setSecurityGroup(String securityGroup) {
    this.securityGroup = securityGroup;
  }

  public void setSecurityID(String securityID) {
    SecurityID = securityID;
  }

  public void setSecurityIDSource(SecurityIDSource securityIDSource) {
    this.securityIDSource = securityIDSource;
  }

  public void setSecurityRejectReason(SecurityRejectReason securityRejectReason) {
    this.securityRejectReason = securityRejectReason;
  }

  public void setSecurityReqID(String securityReqID) {
    this.securityReqID = securityReqID;
  }

  public void setSecurityRequestResult(SecurityRequestResult securityRequestResult) {
    this.securityRequestResult = securityRequestResult;
  }

  public void setSecurityResponseType(SecurityResponseType securityResponseType) {
    this.securityResponseType = securityResponseType;
  }

  public void setSecurityType(String securityType) {
    this.securityType = securityType;
  }

  public void setSettlCurrency(String settlCurrency) {
    this.settlCurrency = settlCurrency;
  }

  public void setSettlPriceType(SettlPriceType settlPriceType) {
    this.settlPriceType = settlPriceType;
  }

  public void setStrikePrice(double strikePrice) {
    this.strikePrice = strikePrice;
  }

  public void setSubscriptionRequestType(SubscriptionRequestType subscriptionRequestType) {
    this.subscriptionRequestType = subscriptionRequestType;
  }

  public void setSymbol(String symbol) {
    this.symbol = symbol;
  }

  public void setTradingSessionID(String tradingSessionID) {
    this.tradingSessionID = tradingSessionID;
  }

  public void setTradingSessionSubID(String tradingSessionSubID) {
    this.tradingSessionSubID = tradingSessionSubID;
  }

  public void setUnderlyingProduct(Product underlyingProduct) {
    this.underlyingProduct = underlyingProduct;
  }
} public enum SecurityIDSource {
  CUSIP('1'), SEDOL('2'), QUIK('3'), ISINNumber('4'), RICCode('5'), ISOCurrencyCode(
      '6'), ISOCountryCode('7'), ExchangeSymbol('8'), ConsolidatedTapeAssociation(
          '9'), BloombergSymbol('A'), Wertpapier('B'), Dutch('C'), Valoren('D'), Sicovam(
              'E'), Belgian('F'), Common('G'), ClearingHouse(
                  'H'), ISDAFpMLSpecification('I'), OptionPriceReportingAuthority('J'), ISDAFpMLURL(
                      'K'), LetterOfCredit('L'), MarketplaceAssignedIdentifier(
                          'M'), MarkitREDEntityCLIP('N'), MarkitREDPairCLIP('P'), CFTCCommodityCode(
                              'Q'), ISDACommodityReferencePrice(
                                  'R'), FinancialInstrumentGlobalIdentifier(
                                      'S'), LegalEntityIdentifier('T'), Synthetic('U');

  private static final EnumSet<SecurityIDSource> valueSet = EnumSet.allOf(SecurityIDSource.class);

  public static SecurityIDSource getValue(char code) {
    Iterator<SecurityIDSource> iter = valueSet.iterator();
    while (iter.hasNext()) {
      SecurityIDSource value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final char code;

  SecurityIDSource(char code) {
    this.code = code;
  }

  public char getCode() {
    return code;
  }
} public enum SecurityRejectReason {
  InvalidInstrumentRequested(1), InstrumentAlreadyExists(2), RequestTypeNotSupported(
      3), SystemUnavailableForInstrumentCreation(4), IneligibleInstrumentGroup(
          5), InstrumentIDUnavailable(6), InvalidOrMissingDataOnOptionLeg(
              7), InvalidOrMissingDataOnFutureLeg(8), InvalidOrMissingDataOnFXLeg(
                  10), InvalidLegPriceSpecified(11), InvalidInstrumentStructureSpecified(12);

  private static final EnumSet<SecurityRejectReason> valueSet = EnumSet.allOf(SecurityRejectReason.class);

  public static SecurityRejectReason getValue(int code) {
    Iterator<SecurityRejectReason> iter = valueSet.iterator();
    while (iter.hasNext()) {
      SecurityRejectReason value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final int code;

  SecurityRejectReason(int code) {
    this.code = code;
  }

  public int getCode() {
    return code;
  }

} public enum SecurityRequestResult {
  ValidRequest(0), InvalidOrUnsupportedRequest(1), NoInstrumentsFound(
      2), NotAuthorizedToRetrieveInstrumentData(
          3), InstrumentDataTemporarilyUnavailable(4), RequestForInstrumentDataNotSupported(5);

  private static final EnumSet<SecurityRequestResult> valueSet =
      EnumSet.allOf(SecurityRequestResult.class);

  public static SecurityRequestResult getValue(int code) {
    Iterator<SecurityRequestResult> iter = valueSet.iterator();
    while (iter.hasNext()) {
      SecurityRequestResult value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final int code;

  SecurityRequestResult(int code) {
    this.code = code;
  }

  public int getCode() {
    return code;
  }

} public enum SecurityResponseType {
  AcceptAsIs(1), AcceptWithRevisions(2), @Deprecated ListOfSecurityTypesReturnedPerRequest(
      3), ListOfSecuritiesReturnedPerRequest(
          4), RejectSecurityProposal(5), CannotMatchSelectionCriteria(6);

  private static final EnumSet<SecurityResponseType> valueSet = EnumSet.allOf(SecurityResponseType.class);

  public static SecurityResponseType getValue(int code) {
    Iterator<SecurityResponseType> iter = valueSet.iterator();
    while (iter.hasNext()) {
      SecurityResponseType value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final int code;

  SecurityResponseType(int code) {
    this.code = code;
  }

  public int getCode() {
    return code;
  }

} public enum SettlPriceType {
  Final(1), Theoretical(2);

  private static final EnumSet<SettlPriceType> valueSet = EnumSet.allOf(SettlPriceType.class);

  public static SettlPriceType getValue(int code) {
    Iterator<SettlPriceType> iter = valueSet.iterator();
    while (iter.hasNext()) {
      SettlPriceType value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final int code;

  SettlPriceType(int code) {
    this.code = code;
  }

  public int getCode() {
    return code;
  }
} public enum SubscriptionRequestType {
  Snapshot('0'), SnapshotAndUpdates('1'), DisablePreviousSnapshot('2');

  private static final EnumSet<SubscriptionRequestType> valueSet =
      EnumSet.allOf(SubscriptionRequestType.class);

  public static SubscriptionRequestType getValue(char code) {
    Iterator<SubscriptionRequestType> iter = valueSet.iterator();
    while (iter.hasNext()) {
      SubscriptionRequestType value = iter.next();
      if (value.getCode() == code) {
        return value;
      }
    }
    return null;
  }

  private final char code;

  SubscriptionRequestType(char code) {
    this.code = code;
  }

  public char getCode() {
    return code;
  }

} public class InstrAttribImpl implements InstrAttrib {

  private InstrAttribType instrAttribType;
  private String instrAttribValue;

  @Override
  public InstrAttribType getInstrAttribType() {
    return instrAttribType;
  }

  @Override
  public String getInstrAttribValue() {
    return instrAttribValue;
  }

  public void setInstrAttribType(InstrAttribType instrAttribType) {
    this.instrAttribType = instrAttribType;
  }

  public void setInstrAttribValue(String instrAttribValue) {
    this.instrAttribValue = instrAttribValue;
  }
} public class MarketSegmentImpl implements MarketSegment {

  private String marketID;
  private String marketSegmentID;


  @Override
  public String getMarketID() {
    return marketID;
  }

  @Override
  public String getMarketSegmentID() {
    return marketSegmentID;
  }


  public void setMarketID(String marketID) {
    this.marketID = marketID;
  }

  public void setMarketSegmentID(String marketSegmentID) {
    this.marketSegmentID = marketSegmentID;
  }
}
import static java.time.temporal.ChronoField.DAY_OF_MONTH;
import static java.time.temporal.ChronoField.HOUR_OF_DAY;
import static java.time.temporal.ChronoField.MICRO_OF_SECOND;
import static java.time.temporal.ChronoField.MINUTE_OF_HOUR;
import static java.time.temporal.ChronoField.MONTH_OF_YEAR;
import static java.time.temporal.ChronoField.SECOND_OF_MINUTE;
import static java.time.temporal.ChronoField.YEAR;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SeekableByteChannel;
import java.nio.file.FileSystems;
import java.nio.file.StandardOpenOption;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.util.Spliterator;
import java.util.function.Consumer;
import org.fixtrading.timpani.codec.TagValueMessageSpliterator;
import org.fixtrading.timpani.codec.TagValueMessageSpliterator.TagValueFieldSpliterator;
import org.fixtrading.timpani.securitydef.datastore.MongoDBSecurityDefinitionStore;
import org.fixtrading.timpani.securitydef.datastore.SecurityDefinitionStore;
import org.fixtrading.timpani.securitydef.messages.InstrAttribType;
import org.fixtrading.timpani.securitydef.messages.Product;
import org.fixtrading.timpani.securitydef.messages.PutOrCall;
import org.fixtrading.timpani.securitydef.messages.SecurityDefinitionImpl;
import org.fixtrading.timpani.securitydef.messages.SecurityIDSource;
import org.fixtrading.timpani.securitydef.messages.SecurityRejectReason;
import org.fixtrading.timpani.securitydef.messages.SecurityRequestResult;
import org.fixtrading.timpani.securitydef.messages.SecurityResponseType;
import org.fixtrading.timpani.securitydef.messages.SettlPriceType;
import org.fixtrading.timpani.securitydef.messages.SubscriptionRequestType;
import org.fixtrading.timpani.securitydef.messages.SecurityDefinitionImpl.InstrAttribImpl;
import org.fixtrading.timpani.securitydef.messages.SecurityDefinitionImpl.MarketSegmentImpl;
package org.fixtrading.timpani.securitydef.datastore.loader;




/**
 * Load security definition store from a file
 * <p>
 * The bulk loader truncates an existing collection before reloading it.
 * MongoDB database name: 'refdata' collection: 'secdef'
 * 
 * @author Don Mendelson
 * 
 * @see <a href="ftp://ftp.cmegroup.com/SBEFix/Production">CME source</a>
 */
public class TagValueSourceBulkLoader {

  /**
   * Run bulk loader
   * @param args first argument is name of file to load
   * @throws Exception if the file or data store cannot be accessed
   */
  public static void main(String[] args) throws Exception {
    if (args.length < 1) {
      System.out.println("Usage TagValueSourceBulkLoader <file-name>");
      System.exit(1);
    }
    try (MongoDBSecurityDefinitionStore store = new MongoDBSecurityDefinitionStore();
        FileChannel channel =
            FileChannel.open(FileSystems.getDefault().getPath(args[0]), StandardOpenOption.READ)) {
      store.open().get();
      store.truncate().get();

      TagValueSourceBulkLoader loader = new TagValueSourceBulkLoader();
      Instant start = Instant.now();
      loader.load(store, channel);
      Instant end = Instant.now();
      long elapsed = Duration.between(start, end).toMillis();
      int insertions = loader.getInsertions();
      double rate = elapsed == 0 ? 0 : insertions * 1000.0 / elapsed;
      System.out.format("%d inserted in %.3f seconds at %.3f insertions/second%n", insertions, 
          elapsed/1000.0, rate);
    }
  }

  private final Consumer<? super ByteBuffer> fieldAction = new Consumer<ByteBuffer>() {
    
    private InstrAttribImpl instrAttribImpl;
    private MarketSegmentImpl marketSegmentImpl;