Revision 42184
Added by Michele Artini about 8 years ago
| MongoPublisherStore.java | ||
|---|---|---|
| 1 | 1 |
package eu.dnetlib.oai.mongo; |
| 2 | 2 |
|
| 3 |
import java.io.ByteArrayOutputStream; |
|
| 3 | 4 |
import java.io.IOException; |
| 4 | 5 |
import java.util.Collection; |
| 5 | 6 |
import java.util.Date; |
| ... | ... | |
| 7 | 8 |
import java.util.concurrent.ArrayBlockingQueue; |
| 8 | 9 |
import java.util.concurrent.BlockingQueue; |
| 9 | 10 |
import java.util.concurrent.TimeUnit; |
| 11 |
import java.util.function.Function; |
|
| 12 |
import java.util.stream.Collectors; |
|
| 10 | 13 |
import java.util.zip.ZipEntry; |
| 11 | 14 |
import java.util.zip.ZipOutputStream; |
| 12 | 15 |
|
| 13 |
import com.google.common.base.Function; |
|
| 16 |
import org.apache.commons.lang3.StringUtils; |
|
| 17 |
import org.apache.commons.logging.Log; |
|
| 18 |
import org.apache.commons.logging.LogFactory; |
|
| 19 |
import org.bson.conversions.Bson; |
|
| 20 |
import org.bson.types.Binary; |
|
| 21 |
|
|
| 14 | 22 |
import com.google.common.base.Predicate; |
| 15 | 23 |
import com.google.common.base.Stopwatch; |
| 16 | 24 |
import com.google.common.collect.Iterables; |
| ... | ... | |
| 30 | 38 |
import com.mongodb.client.model.UpdateOptions; |
| 31 | 39 |
import com.mongodb.client.result.DeleteResult; |
| 32 | 40 |
import com.mongodb.client.result.UpdateResult; |
| 33 |
import eu.dnetlib.data.information.oai.publisher.OaiPublisherRuntimeException; |
|
| 34 |
import eu.dnetlib.data.information.oai.publisher.PublisherField; |
|
| 35 |
import eu.dnetlib.data.information.oai.publisher.conf.OAIConfigurationReader; |
|
| 36 |
import eu.dnetlib.data.information.oai.publisher.info.RecordInfo; |
|
| 37 |
import eu.dnetlib.data.information.oai.publisher.info.SetInfo; |
|
| 38 |
import eu.dnetlib.data.oai.store.PublisherStore; |
|
| 39 |
import eu.dnetlib.data.oai.store.RecordChangeDetector; |
|
| 40 |
import eu.dnetlib.data.oai.store.parser.MongoQueryParser; |
|
| 41 |
import eu.dnetlib.data.oai.store.parser.PublisherRecordParser; |
|
| 42 |
import eu.dnetlib.data.oai.store.sets.MongoSetCollection; |
|
| 43 |
import eu.dnetlib.miscutils.functional.UnaryFunction; |
|
| 44 |
import org.apache.commons.io.output.ByteArrayOutputStream; |
|
| 45 |
import org.apache.commons.lang.StringUtils; |
|
| 46 |
import org.apache.commons.logging.Log; |
|
| 47 |
import org.apache.commons.logging.LogFactory; |
|
| 48 |
import org.bson.conversions.Bson; |
|
| 49 |
import org.bson.types.Binary; |
|
| 50 | 41 |
|
| 42 |
import eu.dnetlib.oai.PublisherField; |
|
| 43 |
import eu.dnetlib.oai.PublisherStore; |
|
| 44 |
import eu.dnetlib.oai.RecordChangeDetector; |
|
| 45 |
import eu.dnetlib.oai.conf.OAIConfigurationReader; |
|
| 46 |
import eu.dnetlib.oai.info.RecordInfo; |
|
| 47 |
import eu.dnetlib.oai.info.SetInfo; |
|
| 48 |
import eu.dnetlib.oai.parser.MongoQueryParser; |
|
| 49 |
import eu.dnetlib.oai.parser.PublisherRecordParser; |
|
| 50 |
import eu.dnetlib.oai.sets.MongoSetCollection; |
|
| 51 |
import eu.dnetlib.rmi.provision.OaiPublisherRuntimeException; |
|
| 52 |
|
|
| 51 | 53 |
public class MongoPublisherStore implements PublisherStore<DNetOAIMongoCursor> {
|
| 52 | 54 |
|
| 53 | 55 |
private static final Log log = LogFactory.getLog(MongoPublisherStore.class); // NOPMD by marko on 11/24/08 5:02 PM |
| ... | ... | |
| 119 | 121 |
|
| 120 | 122 |
@Override |
| 121 | 123 |
public RecordInfo getRecord(final String recordId) {
|
| 122 |
Bson query = Filters.eq(OAIConfigurationReader.ID_FIELD, recordId); |
|
| 123 |
DBObject result = this.collection.find(query).first(); |
|
| 124 |
final Bson query = Filters.eq(OAIConfigurationReader.ID_FIELD, recordId);
|
|
| 125 |
final DBObject result = this.collection.find(query).first();
|
|
| 124 | 126 |
log.debug(result); |
| 125 | 127 |
return this.recordInfoGenerator.transformDBObject(result, true); |
| 126 | 128 |
} |
| 127 | 129 |
|
| 128 | 130 |
@Override |
| 129 |
public RecordInfo getRecord(final String recordId, final UnaryFunction<String, String> unaryFunction) {
|
|
| 130 |
RecordInfo result = this.getRecord(recordId); |
|
| 131 |
public RecordInfo getRecord(final String recordId, final Function<String, String> unaryFunction) {
|
|
| 132 |
final RecordInfo result = this.getRecord(recordId);
|
|
| 131 | 133 |
if (result != null) {
|
| 132 |
String transformedBody = unaryFunction.evaluate(result.getMetadata());
|
|
| 134 |
final String transformedBody = unaryFunction.apply(result.getMetadata());
|
|
| 133 | 135 |
result.setMetadata(transformedBody); |
| 134 | 136 |
} |
| 135 | 137 |
return result; |
| ... | ... | |
| 137 | 139 |
|
| 138 | 140 |
@Override |
| 139 | 141 |
public DNetOAIMongoCursor getRecords(final String queryString, final boolean bodyIncluded, final int limit) {
|
| 140 |
FindIterable<DBObject> iter = loggedFindByQuery(queryString, limit); |
|
| 142 |
final FindIterable<DBObject> iter = loggedFindByQuery(queryString, limit);
|
|
| 141 | 143 |
return new DNetOAIMongoCursor(iter.iterator(), bodyIncluded, this.recordInfoGenerator, this.metadataExtractor); |
| 142 | 144 |
} |
| 143 | 145 |
|
| 144 | 146 |
@Override |
| 145 | 147 |
public DNetOAIMongoCursor getRecords(final String queryString, |
| 146 |
final UnaryFunction<String, String> unaryFunction,
|
|
| 148 |
final Function<String, String> unaryFunction, |
|
| 147 | 149 |
final boolean bodyIncluded, |
| 148 | 150 |
final int limit) {
|
| 149 |
FindIterable<DBObject> iter = loggedFindByQuery(queryString, limit); |
|
| 151 |
final FindIterable<DBObject> iter = loggedFindByQuery(queryString, limit);
|
|
| 150 | 152 |
return new DNetOAIMongoCursor(iter.iterator(), unaryFunction, bodyIncluded, this.recordInfoGenerator, this.metadataExtractor); |
| 151 | 153 |
} |
| 152 | 154 |
|
| 153 | 155 |
private FindIterable<DBObject> loggedFindByQuery(final String queryString, final int limit) {
|
| 154 |
Bson query = this.queryParser.parse(queryString); |
|
| 155 |
long start = System.currentTimeMillis(); |
|
| 156 |
Bson sortByIdAsc = Sorts.orderBy(Sorts.ascending("_id"));
|
|
| 157 |
FindIterable<DBObject> iter = this.collection.find(query).sort(sortByIdAsc).limit(limit); |
|
| 158 |
long end = System.currentTimeMillis(); |
|
| 156 |
final Bson query = this.queryParser.parse(queryString);
|
|
| 157 |
final long start = System.currentTimeMillis();
|
|
| 158 |
final Bson sortByIdAsc = Sorts.orderBy(Sorts.ascending("_id"));
|
|
| 159 |
final FindIterable<DBObject> iter = this.collection.find(query).sort(sortByIdAsc).limit(limit);
|
|
| 160 |
final long end = System.currentTimeMillis();
|
|
| 159 | 161 |
log.debug("Query:" + query + "\ntime to get mongo iterable (ms): " + (end - start));
|
| 160 | 162 |
return iter; |
| 161 | 163 |
} |
| ... | ... | |
| 181 | 183 |
public void ensureIndices() {
|
| 182 | 184 |
final ListIndexesIterable<BasicDBObject> indexesIterable = this.collection.listIndexes(BasicDBObject.class); |
| 183 | 185 |
final IndexOptions indexOptions = new IndexOptions().background(true); |
| 184 |
Stopwatch sw = Stopwatch.createUnstarted(); |
|
| 186 |
final Stopwatch sw = Stopwatch.createUnstarted();
|
|
| 185 | 187 |
sw.start(); |
| 186 | 188 |
// I want to keep the composite indexes that might have been defined manually |
| 187 | 189 |
log.debug("Ensuring currently defined composite indexes:");
|
| 188 |
for (BasicDBObject o : indexesIterable) {
|
|
| 189 |
BasicDBObject fieldIndexed = (BasicDBObject) o.get("key");
|
|
| 190 |
for (final BasicDBObject o : indexesIterable) {
|
|
| 191 |
final BasicDBObject fieldIndexed = (BasicDBObject) o.get("key");
|
|
| 190 | 192 |
if (fieldIndexed.keySet().size() > 1) {
|
| 191 | 193 |
log.debug(o); |
| 192 | 194 |
this.collection.createIndex(fieldIndexed, indexOptions); |
| ... | ... | |
| 194 | 196 |
} |
| 195 | 197 |
|
| 196 | 198 |
// Indexes on single fields. |
| 197 |
for (PublisherField field : this.mongoFields) {
|
|
| 198 |
BasicDBObject mongoIdx = new BasicDBObject(field.getFieldName(), 1); |
|
| 199 |
for (final PublisherField field : this.mongoFields) {
|
|
| 200 |
final BasicDBObject mongoIdx = new BasicDBObject(field.getFieldName(), 1);
|
|
| 199 | 201 |
log.debug("Creating index : " + mongoIdx);
|
| 200 | 202 |
this.collection.createIndex(mongoIdx, indexOptions); |
| 201 | 203 |
} |
| ... | ... | |
| 213 | 215 |
* The creation is performed on the background |
| 214 | 216 |
* </p> |
| 215 | 217 |
* |
| 216 |
* @param fieldNames List of fields to be included in the compound index |
|
| 218 |
* @param fieldNames |
|
| 219 |
* List of fields to be included in the compound index |
|
| 217 | 220 |
* @theStore MongoPublisherStore where to create the index |
| 218 | 221 |
*/ |
| 219 | 222 |
public void createCompoundIndex(final List<String> fieldNames) {
|
| 220 | 223 |
if ((fieldNames == null) || fieldNames.isEmpty()) {
|
| 221 | 224 |
log.fatal("No fields specified for the creation of the compound index");
|
| 222 | 225 |
} |
| 223 |
BasicDBObjectBuilder theIndexBuilder = BasicDBObjectBuilder.start(); |
|
| 224 |
for (String f : fieldNames) {
|
|
| 226 |
final BasicDBObjectBuilder theIndexBuilder = BasicDBObjectBuilder.start();
|
|
| 227 |
for (final String f : fieldNames) {
|
|
| 225 | 228 |
theIndexBuilder.add(f, 1); |
| 226 | 229 |
} |
| 227 |
BasicDBObject theIndex = (BasicDBObject) theIndexBuilder.get(); |
|
| 230 |
final BasicDBObject theIndex = (BasicDBObject) theIndexBuilder.get();
|
|
| 228 | 231 |
log.info("Creating index " + theIndex + " on " + this.getId());
|
| 229 | 232 |
this.getCollection().createIndex(theIndex, new IndexOptions().background(true)); |
| 230 | 233 |
} |
| 231 | 234 |
|
| 232 | 235 |
private void dropDiscarded(final String source) {
|
| 233 | 236 |
if (StringUtils.isBlank(source)) {
|
| 234 |
log.debug("Dropping discarded records from publisherStore " + id);
|
|
| 235 |
discardedCollection.drop(); |
|
| 237 |
log.debug("Dropping discarded records from publisherStore " + this.id);
|
|
| 238 |
this.discardedCollection.drop();
|
|
| 236 | 239 |
} else {
|
| 237 |
log.debug("Dropping discarded records for source " + source + " from publisherStore " + id);
|
|
| 238 |
discardedCollection.deleteMany(Filters.eq(OAIConfigurationReader.SET_FIELD, source)); |
|
| 240 |
log.debug("Dropping discarded records for source " + source + " from publisherStore " + this.id);
|
|
| 241 |
this.discardedCollection.deleteMany(Filters.eq(OAIConfigurationReader.SET_FIELD, source));
|
|
| 239 | 242 |
} |
| 240 | 243 |
} |
| 241 | 244 |
|
| ... | ... | |
| 245 | 248 |
final Object sentinel = new Object(); |
| 246 | 249 |
this.dropDiscarded(source); |
| 247 | 250 |
final Date feedDate = new Date(); |
| 248 |
Thread background = new Thread(new Runnable() {
|
|
| 251 |
final Thread background = new Thread(new Runnable() {
|
|
| 249 | 252 |
|
| 250 | 253 |
@Override |
| 251 | 254 |
public void run() {
|
| 252 |
//For fast feeding we want to use a collection with unack write concern |
|
| 253 |
final MongoCollection<DBObject> unackCollection = collection.withWriteConcern(WriteConcern.UNACKNOWLEDGED); |
|
| 255 |
// For fast feeding we want to use a collection with unack write concern
|
|
| 256 |
final MongoCollection<DBObject> unackCollection = MongoPublisherStore.this.collection.withWriteConcern(WriteConcern.UNACKNOWLEDGED);
|
|
| 254 | 257 |
while (true) {
|
| 255 | 258 |
try {
|
| 256 |
Object record = queue.take(); |
|
| 259 |
final Object record = queue.take();
|
|
| 257 | 260 |
if (record == sentinel) {
|
| 258 | 261 |
break; |
| 259 | 262 |
} |
| 260 | 263 |
safeFeedRecord((String) record, source, feedDate, unackCollection); |
| 261 |
} catch (InterruptedException e) {
|
|
| 264 |
} catch (final InterruptedException e) {
|
|
| 262 | 265 |
log.fatal("got exception in background thread", e);
|
| 263 | 266 |
throw new IllegalStateException(e); |
| 264 | 267 |
} |
| ... | ... | |
| 266 | 269 |
} |
| 267 | 270 |
}); |
| 268 | 271 |
background.start(); |
| 269 |
long startFeed = feedDate.getTime(); |
|
| 272 |
final long startFeed = feedDate.getTime();
|
|
| 270 | 273 |
try {
|
| 271 |
log.info("feeding publisherStore " + id);
|
|
| 274 |
log.info("feeding publisherStore " + this.id);
|
|
| 272 | 275 |
for (final String record : records) {
|
| 273 | 276 |
queue.put(record); |
| 274 | 277 |
} |
| 275 | 278 |
queue.put(sentinel); |
| 276 |
log.info("finished feeding publisherStore " + id);
|
|
| 279 |
log.info("finished feeding publisherStore " + this.id);
|
|
| 277 | 280 |
|
| 278 | 281 |
background.join(); |
| 279 |
} catch (InterruptedException e) {
|
|
| 282 |
} catch (final InterruptedException e) {
|
|
| 280 | 283 |
throw new IllegalStateException(e); |
| 281 | 284 |
} |
| 282 |
long endFeed = System.currentTimeMillis(); |
|
| 283 |
log.fatal("OAI STORE " + id + " FEEDING COMPLETED IN " + (endFeed - startFeed) + "ms");
|
|
| 285 |
final long endFeed = System.currentTimeMillis();
|
|
| 286 |
log.fatal("OAI STORE " + this.id + " FEEDING COMPLETED IN " + (endFeed - startFeed) + "ms");
|
|
| 284 | 287 |
this.setDeletedFlags(feedDate, source); |
| 285 | 288 |
return this.count(); |
| 286 | 289 |
} |
| ... | ... | |
| 297 | 300 |
* @param source |
| 298 | 301 |
*/ |
| 299 | 302 |
private void setDeletedFlags(final Date feedDate, final String source) {
|
| 300 |
//get the collection with ACKNOWLEDGE Write concern |
|
| 301 |
final MongoCollection<DBObject> ackCollection = collection.withWriteConcern(WriteConcern.ACKNOWLEDGED); |
|
| 302 |
Thread deletedSetter = new Thread(new Runnable() {
|
|
| 303 |
// get the collection with ACKNOWLEDGE Write concern
|
|
| 304 |
final MongoCollection<DBObject> ackCollection = this.collection.withWriteConcern(WriteConcern.ACKNOWLEDGED);
|
|
| 305 |
final Thread deletedSetter = new Thread(new Runnable() {
|
|
| 303 | 306 |
|
| 304 | 307 |
@Override |
| 305 | 308 |
public void run() {
|
| ... | ... | |
| 309 | 312 |
filter = Filters.and(filter, Filters.eq(OAIConfigurationReader.SET_FIELD, source)); |
| 310 | 313 |
} |
| 311 | 314 |
log.debug("Delete flag query: " + filter);
|
| 312 |
BasicDBObject update = new BasicDBObject("$set",
|
|
| 315 |
final BasicDBObject update = new BasicDBObject("$set",
|
|
| 313 | 316 |
BasicDBObjectBuilder.start(OAIConfigurationReader.DELETED_FIELD, true).append(OAIConfigurationReader.DATESTAMP_FIELD, feedDate) |
| 314 | 317 |
.append(OAIConfigurationReader.UPDATED_FIELD, true).get()); |
| 315 | 318 |
log.debug("Updating as: " + update.toString());
|
| ... | ... | |
| 321 | 324 |
deletedSetter.start(); |
| 322 | 325 |
try {
|
| 323 | 326 |
deletedSetter.join(); |
| 324 |
} catch (InterruptedException e) {
|
|
| 327 |
} catch (final InterruptedException e) {
|
|
| 325 | 328 |
throw new IllegalStateException(e); |
| 326 | 329 |
} |
| 327 | 330 |
} |
| ... | ... | |
| 333 | 336 |
|
| 334 | 337 |
@Override |
| 335 | 338 |
public void drop(final String queryString) {
|
| 336 |
Bson query = this.queryParser.parse(queryString); |
|
| 339 |
final Bson query = this.queryParser.parse(queryString);
|
|
| 337 | 340 |
final DeleteResult deleteResult = this.collection.deleteMany(query); |
| 338 | 341 |
log.debug("Deleted by query: " + queryString + " #deleted: " + deleteResult.getDeletedCount());
|
| 339 | 342 |
|
| ... | ... | |
| 346 | 349 |
|
| 347 | 350 |
@Override |
| 348 | 351 |
public int count(final String queryString) {
|
| 349 |
if (StringUtils.isBlank(queryString)) return (int) this.collection.count();
|
|
| 350 |
Bson query = this.queryParser.parse(queryString); |
|
| 352 |
if (StringUtils.isBlank(queryString)) { return (int) this.collection.count(); }
|
|
| 353 |
final Bson query = this.queryParser.parse(queryString);
|
|
| 351 | 354 |
return (int) this.collection.count(query); |
| 352 | 355 |
} |
| 353 | 356 |
|
| 354 | 357 |
public List<String> getDistinctSetNamesFromRecords() {
|
| 355 |
log.info("Going to ask for all distinct sets in the oaistore " + id + ": this may take a long time...");
|
|
| 358 |
log.info("Going to ask for all distinct sets in the oaistore " + this.id + ": this may take a long time...");
|
|
| 356 | 359 |
return Lists.newArrayList(this.collection.distinct(OAIConfigurationReader.SET_FIELD, String.class)); |
| 357 | 360 |
} |
| 358 | 361 |
|
| ... | ... | |
| 361 | 364 |
// ***********************************************************************************************// |
| 362 | 365 |
private boolean safeFeedRecord(final String record, final String source, final Date feedDate, final MongoCollection<DBObject> unackCollection) {
|
| 363 | 366 |
try {
|
| 364 |
if (!record.isEmpty()) return feedRecord(record, source, feedDate, unackCollection);
|
|
| 367 |
if (!record.isEmpty()) { return feedRecord(record, source, feedDate, unackCollection); }
|
|
| 365 | 368 |
} catch (final Throwable e) {
|
| 366 | 369 |
log.error("Got unhandled exception while parsing record", e);
|
| 367 |
discardedCollection.insertOne(new BasicDBObject(OAIConfigurationReader.SET_FIELD, source).append(OAIConfigurationReader.BODY_FIELD, record)); |
|
| 370 |
this.discardedCollection.insertOne(new BasicDBObject(OAIConfigurationReader.SET_FIELD, source).append(OAIConfigurationReader.BODY_FIELD, record));
|
|
| 368 | 371 |
} |
| 369 | 372 |
return false; |
| 370 | 373 |
} |
| ... | ... | |
| 375 | 378 |
* @return true if the record is new, false otherwise |
| 376 | 379 |
*/ |
| 377 | 380 |
private boolean feedRecord(final String record, final String source, final Date feedDate, final MongoCollection<DBObject> unackCollection) {
|
| 378 |
PublisherRecordParser parser = new PublisherRecordParser(this.mongoFields); |
|
| 381 |
final PublisherRecordParser parser = new PublisherRecordParser(this.mongoFields);
|
|
| 379 | 382 |
final Multimap<String, String> recordProperties = parser.parseRecord(record); |
| 380 | 383 |
String id = ""; |
| 381 | 384 |
String oaiID = ""; |
| ... | ... | |
| 395 | 398 |
} |
| 396 | 399 |
} else {
|
| 397 | 400 |
log.error("parsed record seems invalid -- no identifier property with name: " + OAIConfigurationReader.ID_FIELD);
|
| 398 |
discardedCollection.insertOne(new BasicDBObject(OAIConfigurationReader.SET_FIELD, source).append(OAIConfigurationReader.BODY_FIELD, record).append( |
|
| 399 |
OAIConfigurationReader.DATESTAMP_FIELD, feedDate)); |
|
| 401 |
this.discardedCollection |
|
| 402 |
.insertOne(new BasicDBObject(OAIConfigurationReader.SET_FIELD, source).append(OAIConfigurationReader.BODY_FIELD, record).append( |
|
| 403 |
OAIConfigurationReader.DATESTAMP_FIELD, feedDate)); |
|
| 400 | 404 |
} |
| 401 | 405 |
return false; |
| 402 | 406 |
} |
| 403 | 407 |
|
| 404 | 408 |
private BasicDBObject createBasicObject(final String oaiID, final String record, final Multimap<String, String> recordProperties) {
|
| 405 |
BasicDBObject obj = new BasicDBObject(); |
|
| 409 |
final BasicDBObject obj = new BasicDBObject();
|
|
| 406 | 410 |
for (final String key : recordProperties.keySet()) {
|
| 407 | 411 |
if (key.equals(OAIConfigurationReader.ID_FIELD)) {
|
| 408 | 412 |
obj.put(key, oaiID); |
| 409 | 413 |
} else {
|
| 410 |
Collection<String> values = recordProperties.get(key); |
|
| 414 |
final Collection<String> values = recordProperties.get(key);
|
|
| 411 | 415 |
if (key.equals(OAIConfigurationReader.SET_FIELD)) {
|
| 412 | 416 |
|
| 413 |
Iterable<String> setSpecs = Iterables.transform(values, new Function<String, String>() {
|
|
| 417 |
final Iterable<String> setSpecs = |
|
| 418 |
values.stream().map(s -> MongoPublisherStore.this.mongoSetCollection.normalizeSetSpec(s)).collect(Collectors.toList()); |
|
| 414 | 419 |
|
| 415 |
@Override |
|
| 416 |
public String apply(final String s) {
|
|
| 417 |
return mongoSetCollection.normalizeSetSpec(s); |
|
| 418 |
} |
|
| 419 |
|
|
| 420 |
}); |
|
| 421 | 420 |
obj.put(key, setSpecs); |
| 422 | 421 |
} else {
|
| 423 | 422 |
// let's check if the key is the name of a repeatable field or not |
| 424 |
PublisherField keyField = Iterables.find(this.mongoFields, new Predicate<PublisherField>() {
|
|
| 423 |
final PublisherField keyField = Iterables.find(this.mongoFields, new Predicate<PublisherField>() {
|
|
| 425 | 424 |
|
| 426 | 425 |
@Override |
| 427 | 426 |
public boolean apply(final PublisherField field) {
|
| ... | ... | |
| 445 | 444 |
obj.put(OAIConfigurationReader.BODY_FIELD, createCompressRecord(record)); |
| 446 | 445 |
obj.put(OAIConfigurationReader.DELETED_FIELD, false); |
| 447 | 446 |
return obj; |
| 448 |
} catch (IOException e) {
|
|
| 447 |
} catch (final IOException e) {
|
|
| 449 | 448 |
throw new OaiPublisherRuntimeException(e); |
| 450 | 449 |
} |
| 451 | 450 |
} |
| ... | ... | |
| 455 | 454 |
* @throws IOException |
| 456 | 455 |
*/ |
| 457 | 456 |
public Binary createCompressRecord(final String record) throws IOException {
|
| 458 |
ByteArrayOutputStream os = new ByteArrayOutputStream(); |
|
| 459 |
ZipOutputStream zos = new ZipOutputStream(os); |
|
| 460 |
ZipEntry entry = new ZipEntry(OAIConfigurationReader.BODY_FIELD); |
|
| 457 |
final ByteArrayOutputStream os = new ByteArrayOutputStream();
|
|
| 458 |
final ZipOutputStream zos = new ZipOutputStream(os);
|
|
| 459 |
final ZipEntry entry = new ZipEntry(OAIConfigurationReader.BODY_FIELD);
|
|
| 461 | 460 |
zos.putNextEntry(entry); |
| 462 | 461 |
zos.write(record.getBytes()); |
| 463 | 462 |
zos.closeEntry(); |
| ... | ... | |
| 472 | 471 |
final Date feedDate, |
| 473 | 472 |
final MongoCollection<DBObject> unackCollection) {
|
| 474 | 473 |
log.debug("New record received. Assigned oai id: " + oaiID);
|
| 475 |
DBObject obj = this.createBasicObject(oaiID, record, recordProperties); |
|
| 474 |
final DBObject obj = this.createBasicObject(oaiID, record, recordProperties);
|
|
| 476 | 475 |
obj.put(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, feedDate); |
| 477 | 476 |
obj.put(OAIConfigurationReader.DATESTAMP_FIELD, feedDate); |
| 478 | 477 |
obj.put(OAIConfigurationReader.UPDATED_FIELD, false); |
| ... | ... | |
| 486 | 485 |
final Date feedDate, |
| 487 | 486 |
final MongoCollection<DBObject> unackCollection) {
|
| 488 | 487 |
log.debug("updating record " + oaiID);
|
| 489 |
BasicDBObject obj = this.createBasicObject(oaiID, record, recordProperties); |
|
| 488 |
final BasicDBObject obj = this.createBasicObject(oaiID, record, recordProperties);
|
|
| 490 | 489 |
obj.put(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, feedDate); |
| 491 | 490 |
obj.put(OAIConfigurationReader.DATESTAMP_FIELD, feedDate); |
| 492 | 491 |
obj.put(OAIConfigurationReader.UPDATED_FIELD, true); |
| 493 |
Bson oldObj = Filters.eq(OAIConfigurationReader.ID_FIELD, oaiID); |
|
| 492 |
final Bson oldObj = Filters.eq(OAIConfigurationReader.ID_FIELD, oaiID);
|
|
| 494 | 493 |
unackCollection.updateOne(oldObj, obj, new UpdateOptions().upsert(true)); |
| 495 | 494 |
this.upsertSets(recordProperties.get(OAIConfigurationReader.SET_FIELD)); |
| 496 | 495 |
} |
| ... | ... | |
| 498 | 497 |
public void upsertSets(final Iterable<String> setNames) {
|
| 499 | 498 |
// feed the list of sets, if any |
| 500 | 499 |
if (setNames != null) {
|
| 501 |
for (String setName : setNames) {
|
|
| 500 |
for (final String setName : setNames) {
|
|
| 502 | 501 |
if (StringUtils.isNotBlank(setName)) {
|
| 503 | 502 |
final SetInfo set = new SetInfo(); |
| 504 |
String setSpec = this.mongoSetCollection.normalizeSetSpec(setName); |
|
| 503 |
final String setSpec = this.mongoSetCollection.normalizeSetSpec(setName);
|
|
| 505 | 504 |
set.setSetSpec(setSpec); |
| 506 | 505 |
set.setSetName(setName); |
| 507 | 506 |
set.setSetDescription("This set contains metadata records whose provenance is " + setName);
|
| 508 | 507 |
set.setEnabled(true); |
| 509 |
String query = "(" + OAIConfigurationReader.SET_FIELD + " = \"" + setSpec + "\") ";
|
|
| 508 |
final String query = "(" + OAIConfigurationReader.SET_FIELD + " = \"" + setSpec + "\") ";
|
|
| 510 | 509 |
set.setQuery(query); |
| 511 | 510 |
this.mongoSetCollection.upsertSet(set, false, getCollection().getNamespace().getDatabaseName()); |
| 512 | 511 |
} |
| ... | ... | |
| 516 | 515 |
|
| 517 | 516 |
private void handleRecord(final String oaiID, final Date lastCollectionDate, final MongoCollection<DBObject> unackCollection) {
|
| 518 | 517 |
log.debug("handling unchanged record " + oaiID);
|
| 519 |
Bson oldObj = Filters.eq(OAIConfigurationReader.ID_FIELD, oaiID); |
|
| 520 |
BasicDBObject update = new BasicDBObject("$set", new BasicDBObject(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, lastCollectionDate));
|
|
| 518 |
final Bson oldObj = Filters.eq(OAIConfigurationReader.ID_FIELD, oaiID);
|
|
| 519 |
final BasicDBObject update = new BasicDBObject("$set", new BasicDBObject(OAIConfigurationReader.LAST_COLLECTION_DATE_FIELD, lastCollectionDate));
|
|
| 521 | 520 |
unackCollection.updateOne(oldObj, update, new UpdateOptions().upsert(true)); |
| 522 | 521 |
} |
| 523 | 522 |
|
| 524 | 523 |
private boolean isNewRecord(final String oaiIdentifier) {
|
| 525 |
if (alwaysNewRecord || (collection.count() == 0)) return true;
|
|
| 524 |
if (this.alwaysNewRecord || (this.collection.count() == 0)) { return true; }
|
|
| 526 | 525 |
return this.collection.find(Filters.eq(OAIConfigurationReader.ID_FIELD, oaiIdentifier)).first() == null; |
| 527 | 526 |
} |
| 528 | 527 |
|
| ... | ... | |
| 531 | 530 |
// ***********************************************************************************************// |
| 532 | 531 |
|
| 533 | 532 |
private boolean isChanged(final String oaiID, final String record) {
|
| 534 |
RecordInfo oldRecord = getRecord(oaiID); |
|
| 535 |
if (oldRecord == null) return StringUtils.isBlank(record);
|
|
| 533 |
final RecordInfo oldRecord = getRecord(oaiID);
|
|
| 534 |
if (oldRecord == null) { return StringUtils.isBlank(record); }
|
|
| 536 | 535 |
return this.recordChangeDetector.differs(oldRecord.getMetadata(), record); |
| 537 | 536 |
} |
| 538 | 537 |
|
| ... | ... | |
| 544 | 543 |
public int hashCode() {
|
| 545 | 544 |
final int prime = 31; |
| 546 | 545 |
int result = 1; |
| 547 |
result = (prime * result) + ((collection == null) ? 0 : collection.hashCode());
|
|
| 548 |
result = (prime * result) + ((id == null) ? 0 : id.hashCode());
|
|
| 549 |
result = (prime * result) + ((interpretation == null) ? 0 : interpretation.hashCode());
|
|
| 550 |
result = (prime * result) + ((layout == null) ? 0 : layout.hashCode());
|
|
| 551 |
result = (prime * result) + ((metadataFormat == null) ? 0 : metadataFormat.hashCode());
|
|
| 546 |
result = (prime * result) + ((this.collection == null) ? 0 : this.collection.hashCode());
|
|
| 547 |
result = (prime * result) + ((this.id == null) ? 0 : this.id.hashCode());
|
|
| 548 |
result = (prime * result) + ((this.interpretation == null) ? 0 : this.interpretation.hashCode());
|
|
| 549 |
result = (prime * result) + ((this.layout == null) ? 0 : this.layout.hashCode());
|
|
| 550 |
result = (prime * result) + ((this.metadataFormat == null) ? 0 : this.metadataFormat.hashCode());
|
|
| 552 | 551 |
return result; |
| 553 | 552 |
} |
| 554 | 553 |
|
| 555 | 554 |
@Override |
| 556 | 555 |
public boolean equals(final Object obj) {
|
| 557 |
if (this == obj) return true;
|
|
| 558 |
if (obj == null) return false;
|
|
| 559 |
if (!(obj instanceof MongoPublisherStore)) return false;
|
|
| 560 |
MongoPublisherStore other = (MongoPublisherStore) obj; |
|
| 561 |
if (collection == null) {
|
|
| 562 |
if (other.collection != null) return false;
|
|
| 563 |
} else if (!collection.equals(other.collection)) return false;
|
|
| 564 |
if (id == null) {
|
|
| 565 |
if (other.id != null) return false;
|
|
| 566 |
} else if (!id.equals(other.id)) return false;
|
|
| 567 |
if (interpretation == null) {
|
|
| 568 |
if (other.interpretation != null) return false;
|
|
| 569 |
} else if (!interpretation.equals(other.interpretation)) return false;
|
|
| 570 |
if (layout == null) {
|
|
| 571 |
if (other.layout != null) return false;
|
|
| 572 |
} else if (!layout.equals(other.layout)) return false;
|
|
| 573 |
if (metadataFormat == null) {
|
|
| 574 |
if (other.metadataFormat != null) return false;
|
|
| 575 |
} else if (!metadataFormat.equals(other.metadataFormat)) return false;
|
|
| 556 |
if (this == obj) { return true; }
|
|
| 557 |
if (obj == null) { return false; }
|
|
| 558 |
if (!(obj instanceof MongoPublisherStore)) { return false; }
|
|
| 559 |
final MongoPublisherStore other = (MongoPublisherStore) obj;
|
|
| 560 |
if (this.collection == null) {
|
|
| 561 |
if (other.collection != null) { return false; }
|
|
| 562 |
} else if (!this.collection.equals(other.collection)) { return false; }
|
|
| 563 |
if (this.id == null) {
|
|
| 564 |
if (other.id != null) { return false; }
|
|
| 565 |
} else if (!this.id.equals(other.id)) { return false; }
|
|
| 566 |
if (this.interpretation == null) {
|
|
| 567 |
if (other.interpretation != null) { return false; }
|
|
| 568 |
} else if (!this.interpretation.equals(other.interpretation)) { return false; }
|
|
| 569 |
if (this.layout == null) {
|
|
| 570 |
if (other.layout != null) { return false; }
|
|
| 571 |
} else if (!this.layout.equals(other.layout)) { return false; }
|
|
| 572 |
if (this.metadataFormat == null) {
|
|
| 573 |
if (other.metadataFormat != null) { return false; }
|
|
| 574 |
} else if (!this.metadataFormat.equals(other.metadataFormat)) { return false; }
|
|
| 576 | 575 |
return true; |
| 577 | 576 |
} |
| 578 | 577 |
|
| 579 | 578 |
public MongoCollection<DBObject> getCollection() {
|
| 580 |
return collection; |
|
| 579 |
return this.collection;
|
|
| 581 | 580 |
} |
| 582 | 581 |
|
| 583 | 582 |
public void setCollection(final MongoCollection<DBObject> collection) {
|
| ... | ... | |
| 585 | 584 |
} |
| 586 | 585 |
|
| 587 | 586 |
public MongoQueryParser getQueryParser() {
|
| 588 |
return queryParser; |
|
| 587 |
return this.queryParser;
|
|
| 589 | 588 |
} |
| 590 | 589 |
|
| 591 | 590 |
public void setQueryParser(final MongoQueryParser queryParser) {
|
| ... | ... | |
| 593 | 592 |
} |
| 594 | 593 |
|
| 595 | 594 |
public MongoCollection<DBObject> getDiscardedCollection() {
|
| 596 |
return discardedCollection; |
|
| 595 |
return this.discardedCollection;
|
|
| 597 | 596 |
} |
| 598 | 597 |
|
| 599 | 598 |
public void setDiscardedCollection(final MongoCollection<DBObject> discardedCollection) {
|
| ... | ... | |
| 601 | 600 |
} |
| 602 | 601 |
|
| 603 | 602 |
public String getIdScheme() {
|
| 604 |
return idScheme; |
|
| 603 |
return this.idScheme;
|
|
| 605 | 604 |
} |
| 606 | 605 |
|
| 607 | 606 |
public void setIdScheme(final String idScheme) {
|
| ... | ... | |
| 609 | 608 |
} |
| 610 | 609 |
|
| 611 | 610 |
public String getIdNamespace() {
|
| 612 |
return idNamespace; |
|
| 611 |
return this.idNamespace;
|
|
| 613 | 612 |
} |
| 614 | 613 |
|
| 615 | 614 |
public void setIdNamespace(final String idNamespace) {
|
| ... | ... | |
| 617 | 616 |
} |
| 618 | 617 |
|
| 619 | 618 |
public RecordInfoGenerator getRecordInfoGenerator() {
|
| 620 |
return recordInfoGenerator; |
|
| 619 |
return this.recordInfoGenerator;
|
|
| 621 | 620 |
} |
| 622 | 621 |
|
| 623 | 622 |
public void setRecordInfoGenerator(final RecordInfoGenerator recordInfoGenerator) {
|
| ... | ... | |
| 625 | 624 |
} |
| 626 | 625 |
|
| 627 | 626 |
public MetadataExtractor getMetadataExtractor() {
|
| 628 |
return metadataExtractor; |
|
| 627 |
return this.metadataExtractor;
|
|
| 629 | 628 |
} |
| 630 | 629 |
|
| 631 | 630 |
public void setMetadataExtractor(final MetadataExtractor metadataExtractor) {
|
| ... | ... | |
| 633 | 632 |
} |
| 634 | 633 |
|
| 635 | 634 |
public RecordChangeDetector getRecordChangeDetector() {
|
| 636 |
return recordChangeDetector; |
|
| 635 |
return this.recordChangeDetector;
|
|
| 637 | 636 |
} |
| 638 | 637 |
|
| 639 | 638 |
public void setRecordChangeDetector(final RecordChangeDetector recordChangeDetector) {
|
| ... | ... | |
| 677 | 676 |
} |
| 678 | 677 |
|
| 679 | 678 |
public MongoSetCollection getMongoSetCollection() {
|
| 680 |
return mongoSetCollection; |
|
| 679 |
return this.mongoSetCollection;
|
|
| 681 | 680 |
} |
| 682 | 681 |
|
| 683 | 682 |
public void setMongoSetCollection(final MongoSetCollection mongoSetCollection) {
|
| ... | ... | |
| 685 | 684 |
} |
| 686 | 685 |
|
| 687 | 686 |
public List<PublisherField> getMongoFields() {
|
| 688 |
return mongoFields; |
|
| 687 |
return this.mongoFields;
|
|
| 689 | 688 |
} |
| 690 | 689 |
|
| 691 | 690 |
public void setMongoFields(final List<PublisherField> mongoFields) {
|
| ... | ... | |
| 693 | 692 |
} |
| 694 | 693 |
|
| 695 | 694 |
public boolean isAlwaysNewRecord() {
|
| 696 |
return alwaysNewRecord; |
|
| 695 |
return this.alwaysNewRecord;
|
|
| 697 | 696 |
} |
| 698 | 697 |
|
| 699 | 698 |
public void setAlwaysNewRecord(final boolean alwaysNewRecord) {
|
Also available in: Unified diff
oai import