Revision 45479
Added by Alessia Bardi about 7 years ago
| MongoPublisherStore.java | ||
|---|---|---|
| 32 | 32 |
import com.mongodb.client.result.DeleteResult; |
| 33 | 33 |
import com.mongodb.client.result.UpdateResult; |
| 34 | 34 |
import eu.dnetlib.cql.CqlTranslator; |
| 35 |
import eu.dnetlib.cql.mongo.MongoCqlTranslator; |
|
| 36 | 35 |
import eu.dnetlib.oai.PublisherField; |
| 37 | 36 |
import eu.dnetlib.oai.PublisherStore; |
| 38 | 37 |
import eu.dnetlib.oai.RecordChangeDetector; |
| ... | ... | |
| 246 | 245 |
final Object sentinel = new Object(); |
| 247 | 246 |
this.dropDiscarded(source); |
| 248 | 247 |
final Date feedDate = new Date(); |
| 249 |
final Thread background = new Thread(new Runnable() {
|
|
| 250 |
|
|
| 251 |
@Override |
|
| 252 |
public void run() {
|
|
| 253 |
// For fast feeding we want to use a collection with unack write concern |
|
| 254 |
final MongoCollection<DBObject> unackCollection = MongoPublisherStore.this.collection.withWriteConcern(WriteConcern.UNACKNOWLEDGED); |
|
| 255 |
while (true) {
|
|
| 256 |
try {
|
|
| 257 |
final Object record = queue.take(); |
|
| 258 |
if (record == sentinel) {
|
|
| 259 |
break; |
|
| 260 |
} |
|
| 261 |
safeFeedRecord((String) record, source, feedDate, unackCollection); |
|
| 262 |
} catch (final InterruptedException e) {
|
|
| 263 |
log.fatal("got exception in background thread", e);
|
|
| 264 |
throw new IllegalStateException(e); |
|
| 248 |
final Thread background = new Thread(() -> {
|
|
| 249 |
// For fast feeding we want to use a collection with unack write concern |
|
| 250 |
final MongoCollection<DBObject> unackCollection = MongoPublisherStore.this.collection.withWriteConcern(WriteConcern.UNACKNOWLEDGED); |
|
| 251 |
while (true) {
|
|
| 252 |
try {
|
|
| 253 |
final Object record = queue.take(); |
|
| 254 |
if (record == sentinel) {
|
|
| 255 |
break; |
|
| 265 | 256 |
} |
| 257 |
safeFeedRecord((String) record, source, feedDate, unackCollection); |
|
| 258 |
} catch (final InterruptedException e) {
|
|
| 259 |
log.fatal("got exception in background thread", e);
|
|
| 260 |
throw new IllegalStateException(e); |
|
| 266 | 261 |
} |
| 267 | 262 |
} |
| 268 | 263 |
}); |
| ... | ... | |
| 380 | 375 |
*/ |
| 381 | 376 |
private boolean feedRecord(final String record, final String source, final Date feedDate, final MongoCollection<DBObject> unackCollection) {
|
| 382 | 377 |
final PublisherRecordParser parser = new PublisherRecordParser(this.mongoFields); |
| 383 |
final Multimap<String, String> recordProperties = parser.parseRecord(record); |
|
| 378 |
log.debug("configured parser for fields: "+this.mongoFields);
|
|
| 379 |
final Multimap<String, String> recordProperties = parser.parseRecord(record, source); |
|
| 384 | 380 |
String id = ""; |
| 385 | 381 |
String oaiID = ""; |
| 386 | 382 |
if (recordProperties.containsKey(OAIConfigurationReader.ID_FIELD)) {
|
| ... | ... | |
| 399 | 395 |
} |
| 400 | 396 |
} else {
|
| 401 | 397 |
log.error("parsed record seems invalid -- no identifier property with name: " + OAIConfigurationReader.ID_FIELD);
|
| 398 |
log.debug("Extracted property map: \n"+recordProperties);
|
|
| 399 |
log.debug("from: \n"+record);
|
|
| 402 | 400 |
this.discardedCollection |
| 403 | 401 |
.insertOne(new BasicDBObject(OAIConfigurationReader.SET_FIELD, source).append(OAIConfigurationReader.BODY_FIELD, record).append( |
| 404 | 402 |
OAIConfigurationReader.DATESTAMP_FIELD, feedDate)); |
Also available in: Unified diff
More logs for debug