import oracle.kv.Value; //导入依赖的package包/类
/**
 * Store the body of the event as the value of the key. The assumption
 * is that the event body will already be serialized as an Avro record
 * or in another suitable format.
 *
 * @param tableName the name of the table
 * @param avroBody the encoded event body
 * @return the body of the event as a Value object.
 */
protected Value getValue(String tableName, byte[] avroBody) {
    RawRecord raw;
    Schema schema;
    
    LOG.debug("getValue(): tableName={}", tableName);

    
    schema = schemas.get(tableName);
    if (schema == null) {
        throw new RuntimeException("Avro schema not found: " + tableName);
    }
    raw = new RawRecord(avroBody, schema);
    
    return binding.toValue(raw);
}
 

import oracle.kv.Value; //导入依赖的package包/类
public String toOracleLoaderFormat(KeyValueVersion kvv,
                                   KVStore kvStore) {
    final Value value = kvv.getValue();

    String movieJsonTxt = null;
    MovieTO movieTO = null;
    String returnStr = null;

    if (value!=null) {
        movieTO = movieDAO.getMovieTO(value);
        
        returnStr = movieTO.toString();
    }

    return returnStr;
}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method returns CrewTO when crewId is passed
 * @param crewId uniue Cast Id
 * @return CrewTO
 */
public CrewTO getCrewById(int crewId) {
    CrewTO crewTO = null;
    ValueVersion vv = null;
    Value crewTOValue = null;

    if (crewId > 0) {

        /**
     * Insert cast json string to the crewId
     * Key=/CW/crewId/-/info
     */
        Key key = KeyUtil.getCrewInfoKey(crewId);
        vv = getKVStore().get(key);

        if (vv != null) {
            crewTOValue = vv.getValue();
            crewTO = this.getCrewTO(crewTOValue);
        } //if(StringUtil.isNotEmpty(jsonTxt))
    } //if (castId > 0)
    return crewTO;
}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes CrewTO object and serialize it to Value object.
 * @param crewTO
 * @return Value
 */
public Value toValue(CrewTO crewTO) {
    ObjectNode crewNode = null;
    JsonRecord jsonRecord = null;
    Value value = null;

    if (crewTO != null) {
        crewNode = crewTO.getCrewJson();
        jsonRecord = new JsonRecord(crewNode, crewSchema);
        // serialize CrewTO to byte array using JSONAvroBinding
        value = crewBinding.toValue(jsonRecord);
    }

    return value;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes Value object as input and deserialize it into CrewTO. The assumption here is that
 * Value that is passed as argument is serialized CrewTO JSON object.
 * @param crewTOValue
 * @return CrewTO
 */
public CrewTO getCrewTO(Value crewTOValue) {

    CrewTO crewTO = null;
    ObjectNode crewNode = null;
    JsonRecord jsonRecord = null;

    if (crewTOValue != null) {
        jsonRecord = crewBinding.toObject(crewTOValue);
        crewNode = (ObjectNode)jsonRecord.getJsonNode();
        crewTO = new CrewTO(crewNode);
    }

    return crewTO;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method returns CastTO when castId is passed
 * @param castId uniue Cast Id
 * @return CastTO
 */
public CastTO getCastById(int castId) {
    CastTO castTO = null;
    Value castTOValue = null;

    if (castId > 0) {
        /**
     * Insert cast json string to the castId
     * Key=/CT/castId/-/info
     */
        Key key = KeyUtil.getCastInfoKey(castId);
        ValueVersion vv = getKVStore().get(key);
        if (vv != null) {
            castTOValue = vv.getValue();
            castTO = this.getCastTO(castTOValue);

        } //if (vv != null)

    } //if (castId > 0)
    return castTO;
}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes CastTO object and serialize it to Value object.
 * @param castTO
 * @return Value
 */
protected Value toValue(CastTO castTO) {
    ObjectNode castNode = null;
    JsonRecord jsonRecord = null;
    Value value = null;

    if (castTO != null) {
        castNode = castTO.geCastJson();
        jsonRecord = new JsonRecord(castNode, castSchema);
        // serialize CastTO to byte array using JSONAvroBinding
        value = castBinding.toValue(jsonRecord);
    }

    return value;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes Value object as input and deserialize it into CastTO. The assumption here is that
 * Value that is passed as argument is serialized CastTO JSON object.
 * @param castTOValue
 * @return CastTO
 */
protected CastTO getCastTO(Value castTOValue) {

    CastTO castTO = null;
    ObjectNode castNode = null;
    JsonRecord jsonRecord = null;

    if (castTOValue != null) {
        jsonRecord = castBinding.toObject(castTOValue);
        castNode = (ObjectNode)jsonRecord.getJsonNode();
        castTO = new CastTO(castNode);
    }

    return castTO;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method returns all different Genres available in the store
 * @return List of GenreTO
 */
public List<GenreTO> getGenres() {

    KeyValueVersion keyValue = null;
    Value genreTOValue = null;        
    GenreTO genreTO = null;

    List<GenreTO> genreList = new ArrayList<GenreTO>();
    Key key = Key.createKey(KeyConstant.GENRE_TABLE);
    Iterator<KeyValueVersion> keyValueIter = getKVStore().storeIterator(Direction.UNORDERED, 0, key, null, null);

    while (keyValueIter.hasNext()) {
        keyValue = keyValueIter.next();
        genreTOValue = keyValue.getValue();
        
        genreTO = this.getGenreTO(genreTOValue);
        //Add TO to the list
        genreList.add(genreTO);

        //System.out.println(genreTO.getId() + " " + genreTO.getGenreJsonTxt());
    } //EOF while

    return genreList;
}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method returns GenreTO for the genreName
 * @param genreName
 * @return GenreTO
 */
public GenreTO getGenreByName(String genreName) {

    GenreTO genreTO = null;
    Value genreTOValue = null;

    if (StringUtil.isNotEmpty(genreName)) {
        Key key = KeyUtil.getGenreNameKey(genreName);
        ValueVersion vv = getKVStore().get(key);
        if (vv != null) {
            genreTOValue = vv.getValue();
            genreTO = this.getGenreTO(genreTOValue);

        } //if (key != null)
    }
    return genreTO;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes CustomerTO object and serialize it to Value object.
 * @param genreTO
 * @return Value
 */
public Value toValue(GenreTO genreTO) {
    ObjectNode genreNode = null;
    JsonRecord jsonRecord = null;
    Value value = null;

    if (genreTO != null) {
        genreNode = genreTO.getGenreJson();
        jsonRecord = new JsonRecord(genreNode, genreSchema);
        // serialize CustomerTO to byte array using JSONAvroBinding
        value = genreBinding.toValue(jsonRecord);
    }

    return value;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes Value object as input and deserialize it into GenreTO. The assumption here is that
 * Value that is passed as argument is serialized GenreTO JSON object.
 * @param genreTOValue
 * @return GenreTO
 */
public GenreTO getGenreTO(Value genreTOValue) {

    GenreTO genreTO = null;
    ObjectNode genreNode = null;
    JsonRecord jsonRecord = null;

    if (genreTOValue != null) {
        jsonRecord = genreBinding.toObject(genreTOValue);
        genreNode = (ObjectNode)jsonRecord.getJsonNode();
        genreTO = new GenreTO(genreNode);
    }

    return genreTO;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method inserts customer profile information into the store if it
 * does not exist already.
 * @param customerTO contains customer's name, email, username, password &
 * customer ID information.
 * @return Version of the new value, or null if an existing value is present
 * and the put is unsuccessful.
 */
public Version insertCustomerProfile(CustomerTO customerTO, boolean force) {

    Key key = null;
    Value value = null;
    Version version = null;
   
    if (customerTO != null) {            
        // key=/CUST/userName
        key = KeyUtil.getCustomerKey(customerTO.getUserName());
        // serialize CustomerTO to byte array using JSONAvroBinding
        value = this.toValue(customerTO);

        if (force) {
            //Insert profile no matter it already existed
            version = getKVStore().put(key, value);
        } else {
            //Insert if profile by same userName & password doesn't exist
            version = getKVStore().putIfAbsent(key, value);
        }
    } //if(customerTO!=null){
    return version;
}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes CustomerTO object and serialize it to Value object.
 * @param customerTO
 * @return
 */
public Value toValue(CustomerTO customerTO){
    ObjectNode customerNode = null;
    JsonRecord jsonRecord = null;
    Value value = null;
    
    if (customerTO != null) {
        customerNode = customerTO.geCustomertJson();
        jsonRecord = new JsonRecord(customerNode, customerSchema);            
        // serialize CustomerTO to byte array using JSONAvroBinding
        value = customerBinding.toValue(jsonRecord);
    }
    
    return value;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes Value object as input and deserialize it into CustomerTO. The assumption here is that
 * Value that is passed as argument is serialized CustomerTO JSON object.
 * @param custTOValue
 * @return CustomerTO
 */
public CustomerTO getCustomerTO(Value custTOValue){
    
    CustomerTO custTO = null;
    ObjectNode customerNode = null;
    JsonRecord jsonRecord = null;
    
    if(custTOValue!=null){
        jsonRecord = customerBinding.toObject(custTOValue);
        customerNode = (ObjectNode) jsonRecord.getJsonNode();
        custTO = new CustomerTO(customerNode);
    }
    
    return custTO; 
    
}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * Movie names are also indexed while inserting the movie data therefore
 * movie can be fetched my full movie name (case insensitive).
 * @param movieName
 * @return List of MovieTO
 */
public List<MovieTO> getMoviesByName(String movieName) {
    Key key = KeyUtil.getMovieNameKey(movieName, 0);
    List<MovieTO> movieTOList = new ArrayList<MovieTO>();
    MovieTO movieTO = null;
    KeyValueVersion keyValue = null;
    Value value = null;
    String movieIdStr = null;

    Iterator<KeyValueVersion> keyIter = super.getKVStore().multiGetIterator(Direction.FORWARD, 0,
            /*key*/key, null, null);
    while (keyIter.hasNext()) {
        keyValue = keyIter.next();
        value = keyValue.getValue();
        movieIdStr = new String(value.getValue());
        movieTO = getMovieById(movieIdStr);

        //add the movie to the list
        movieTOList.add(movieTO);
        //System.out.println(keyValue.getKey().getFullPath() + " \n" +  movieTO.getMovieJson() + "\n ");
    } //EOF while

    return movieTOList;
}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method returns a list of all the movie titles stored in the database
 * @return List of MovieTO
 */
public List<MovieTO> getMovies() {

    Key key = key = Key.createKey(KeyConstant.MOVIE_TABLE);
    KeyValueVersion keyValue = null;
    Value value = null;
    MovieTO movieTO = null;
    List<MovieTO> movieList = new ArrayList<MovieTO>();

    Iterator<KeyValueVersion> keyIter = getKVStore().storeIterator(Direction.UNORDERED, 0,
            /*key*/key, null, null);
    while (keyIter.hasNext()) {
        keyValue = keyIter.next();
        value = keyValue.getValue();
        
        if (value!=null) {
            movieTO = this.getMovieTO(value);
            
            if (movieTO != null) {
                movieList.add(movieTO);
            }
        }

    } //EOF while
    return movieList;
}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes CustomerTO object and serialize it to Value object.
 * @param movieTO
 * @return
 */
public Value toValue(MovieTO movieTO) {
    ObjectNode movieNode = null;
    JsonRecord jsonRecord = null;
    Value value = null;

    if (movieTO != null) {
        movieNode = movieTO.getMovieJson();
        jsonRecord = new JsonRecord(movieNode, movieSchema);
        // serialize CustomerTO to byte array using JSONAvroBinding
        value = movieBinding.toValue(jsonRecord);
    }

    return value;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method takes Value object as input and deserialize it into CustomerTO. The assumption here is that
 * Value that is passed as argument is serialized CustomerTO JSON object.
 * @param movieTOValue
 * @return CustomerTO
 */
public MovieTO getMovieTO(Value movieTOValue) {

    MovieTO movieTO = null;
    ObjectNode movieNode = null;
    JsonRecord jsonRecord = null;

    if (movieTOValue != null) {
        jsonRecord = movieBinding.toObject(movieTOValue);
        movieNode = (ObjectNode)jsonRecord.getJsonNode();
        movieTO = new MovieTO(movieNode);
    }

    return movieTO;

}
 

import oracle.kv.Value; //导入依赖的package包/类
/**
 * This method returns the ActivityTO for the movie saved in the Currently
 * watching list. When a user pause a movie they were watching the position
 * of the movie is saved as ActivityTO, so if you want to find out what
 * position the movie was paused at you need to get the object from the same
 * currently watching list.
 * If ActivityTO is null that means movie has not been started yet.
 * @param custId
 * @param movieId
 * @return ActivityTO
 */
public ActivityTO getActivityTO(int custId, int movieId) {
    ActivityTO activityTO = null;
    Value value = null;
    ValueVersion vv = null;
    Key key = null;
    
    //update the current position of the movie into current watch list
    key = KeyUtil.getCustomerCurrentWatchListKey(custId, movieId);
    vv = this.getKVStore().get(key);

    //System.out.println("Position in the jsonTxt: " + jsonTxt);

    if (vv!=null) {
        value = vv.getValue();
        activityTO = this.getActivityTO(value);
    }

    return activityTO;
}