@ -1,29 +1,30 @@
package org.gcube.data.access.storagehub.handlers.vres ;
import java.text.SimpleDateFormat ;
import java.util.ArrayList ;
import java.util.Arrays ;
import java.util.Calendar ;
import java.util.Iterator ;
import java.util.Collections ;
import java.util.HashMap ;
import java.util.LinkedList ;
import java.util.List ;
import java.util.Locale ;
import java.util. Map ;
import java.util.concurrent.Callable ;
import java.util.stream.Collectors ;
import javax.jcr.Credentials ;
import javax.jcr.Node ;
import javax.jcr.NodeIterator ;
import javax.jcr.Property ;
import javax.jcr.Repository ;
import javax.jcr.RepositoryException ;
import javax.jcr.Session ;
import javax.jcr.observation.Event ;
import javax.jcr.observation.EventJournal ;
import javax.jcr.query.Query ;
import org.apache.jackrabbit.util.ISO9075 ;
import org.gcube.common.storagehub.model.Excludes ;
import org.gcube.common.storagehub.model.NodeConstants ;
import org.gcube.common.storagehub.model.exceptions.BackendGenericError ;
import org.gcube.common.storagehub.model.items.AbstractFileItem ;
import org.gcube.common.storagehub.model.items.FolderItem ;
import org.gcube.common.storagehub.model.items.Item ;
import org.gcube.data.access.storagehub.NodeChildrenFilterIterator ;
import org.gcube.data.access.storagehub.handlers.items.Node2ItemConverter ;
import org.slf4j.Logger ;
import org.slf4j.LoggerFactory ;
@ -37,92 +38,154 @@ public class VREQueryRetriever implements Callable<List<Item>> {
private Repository repository ;
private Credentials credentials ;
private Item vreFolder ;
List < Item > cachedList = new ArrayList < > ( CACHE_DIMENSION ) ;
Map < String , Long > cachedMap = new HashMap < > ( CACHE_DIMENSION ) ;
long higherTimestamp = Long . MAX_VALUE ;
long lastTimestamp = 0 ;
private Node2ItemConverter node2Item = new Node2ItemConverter ( ) ;
long doTime = 0 ;
boolean underRedo = false ;
Node2ItemConverter node2Item ;
public VREQueryRetriever ( Repository repository , Credentials credentials , Item vreFolder ) {
public VREQueryRetriever ( Repository repository , Credentials credentials , Node2ItemConverter node2Item , Item vreFolder ) {
super ( ) ;
this . repository = repository ;
this . credentials = credentials ;
this . vreFolder = vreFolder ;
this . node2Item = node2Item ;
}
public List < Item > call ( ) {
logger . trace ( "executing recents task" ) ;
Session ses = null ;
if ( lastTimestamp = = 0 ) {
try {
long start = System . currentTimeMillis ( ) ;
ses = repository . login ( credentials ) ;
Calendar now = Calendar . getInstance ( ) ;
now . add ( Calendar . YEAR , - 1 ) ;
SimpleDateFormat formatter = new SimpleDateFormat ( "yyyy-MM-dd'T'HH:mm:ss.SSS" , Locale . ENGLISH ) ;
String formattedDate = formatter . format ( now . getTime ( ) ) ;
String xpath = String . format ( "/jcr:root%s//element(*,nthl:workspaceLeafItem)[@jcr:lastModified>xs:dateTime('%s')] order by @jcr:lastModified descending" , ISO9075 . encodePath ( vreFolder . getPath ( ) ) , formattedDate ) ;
//String query = String.format("SELECT * FROM [nthl:workspaceLeafItem] AS node WHERE ISDESCENDANTNODE('%s') ORDER BY node.[jcr:lastModified] DESC ",vreFolder.getPath());
logger . debug ( "query for recents is {}" , xpath ) ;
Query jcrQuery = ses . getWorkspace ( ) . getQueryManager ( ) . createQuery ( xpath , Query . XPATH ) ;
jcrQuery . setLimit ( CACHE_DIMENSION ) ;
lastTimestamp = System . currentTimeMillis ( ) ;
NodeChildrenFilterIterator it = new NodeChildrenFilterIterator ( jcrQuery . execute ( ) . getNodes ( ) ) ;
logger . debug ( "query for recents took {}" , System . currentTimeMillis ( ) - start ) ;
while ( it . hasNext ( ) ) {
Node node = it . next ( ) ;
Item item = node2Item . getItem ( node , Excludes . EXCLUDE_ACCOUNTING ) ;
logger . debug ( "RECENTS - adding item {} with timestamp {}" , item . getTitle ( ) , item . getLastModificationTime ( ) . getTimeInMillis ( ) ) ;
cachedList . add ( item ) ;
}
logger . debug ( "creating objects took {}" , System . currentTimeMillis ( ) - start ) ;
if ( cachedList . size ( ) < = 10 ) return cachedList ;
else return cachedList . subList ( 0 , 10 ) ;
} catch ( Exception e ) {
logger . error ( "error querying vre {}" , vreFolder . getTitle ( ) , e ) ;
throw new RuntimeException ( e ) ;
} finally {
if ( ses ! = null )
ses . logout ( ) ;
logger . trace ( "recents task finished" ) ;
try {
ses = repository . login ( credentials ) ;
Calendar now = Calendar . getInstance ( ) ;
now . add ( Calendar . HOUR_OF_DAY , - 1 ) ;
if ( doTime > now . getTimeInMillis ( ) | | underRedo ) {
logger . debug ( "executing recents task for {} (cahced result)" , vreFolder . getTitle ( ) ) ;
return correctValues ( ses , cachedMap ) ;
} else {
logger . debug ( "executing recents task for {} (redoing it)" , vreFolder . getTitle ( ) ) ;
List < Item > toReturn = redo ( ses ) ;
doTime = System . currentTimeMillis ( ) ;
return toReturn ;
}
} else {
} catch ( Exception e ) {
logger . error ( "error preparing recents for folder {}" , vreFolder . getTitle ( ) , e ) ;
return Collections . emptyList ( ) ;
} finally {
if ( ses ! = null )
ses . logout ( ) ;
logger . debug ( "recents task finished" ) ;
}
}
public synchronized List < Item > redo ( Session ses ) {
underRedo = true ;
try {
Map < String , Long > tempCachedMap = new HashMap < > ( CACHE_DIMENSION ) ;
if ( cachedMap . isEmpty ( ) )
init ( ses , tempCachedMap ) ;
else {
logger . debug ( "redoing recents for {}" , vreFolder . getTitle ( ) ) ;
update ( ses , tempCachedMap ) ;
}
cachedMap = tempCachedMap ;
return correctValues ( ses , tempCachedMap ) ;
} finally {
underRedo = false ;
}
}
private List < Item > correctValues ( Session ses , Map < String , Long > tempCachedMap ) {
logger . debug ( "preparing returning values for {}" , vreFolder . getTitle ( ) ) ;
long start = System . currentTimeMillis ( ) ;
List < Map . Entry < String , Long > > list = new LinkedList < > ( tempCachedMap . entrySet ( ) ) ;
list . sort ( ( c1 , c2 ) - > c1 . getValue ( ) . compareTo ( c2 . getValue ( ) ) * - 1 ) ;
if ( list . size ( ) > CACHE_DIMENSION )
for ( int index = CACHE_DIMENSION - 1 ; index < list . size ( ) ; index + + )
tempCachedMap . remove ( list . get ( index ) . getKey ( ) ) ;
List < String > cachedIds = list . stream ( ) . map ( m - > m . getKey ( ) ) . collect ( Collectors . toList ( ) ) ;
if ( cachedIds . size ( ) > 10 )
cachedIds = cachedIds . subList ( 0 , 10 ) ;
List < Item > result = new ArrayList < > ( 10 ) ;
for ( String id : cachedIds ) {
try {
Item item = node2Item . getItem ( id , ses , Excludes . EXCLUDE_ACCOUNTING ) ;
if ( item ! = null )
result . add ( item ) ;
else logger . warn ( "item with id {} is null" , id ) ;
} catch ( BackendGenericError | RepositoryException e ) { }
}
logger . debug ( "returning values prepared in {} for {}" , System . currentTimeMillis ( ) - start , vreFolder . getTitle ( ) ) ;
return result ;
}
private void insertItemInTheRightPlace ( Item item , Map < String , Long > tempCachedMap ) {
long lastModifiedTime = item . getLastModificationTime ( ) . getTime ( ) . getTime ( ) ;
if ( ! ( lastModifiedTime > higherTimestamp & & tempCachedMap . size ( ) > CACHE_DIMENSION ) ) {
tempCachedMap . put ( item . getId ( ) , lastModifiedTime ) ;
if ( lastModifiedTime > higherTimestamp ) higherTimestamp = lastModifiedTime ;
}
}
private void init ( Session ses , Map < String , Long > tempCachedMap ) {
try {
long start = System . currentTimeMillis ( ) ;
lastTimestamp = System . currentTimeMillis ( ) ;
Node vreFolderNode = ses . getNodeByIdentifier ( vreFolder . getId ( ) ) ;
logger . debug ( "starting visiting children for {}" , vreFolder . getTitle ( ) ) ;
visitChildren ( vreFolderNode , tempCachedMap ) ;
logger . debug ( "initializing recents for {} took {}" , vreFolder . getTitle ( ) , System . currentTimeMillis ( ) - start ) ;
} catch ( Exception e ) {
logger . error ( "error querying vre {}" , vreFolder . getTitle ( ) , e ) ;
throw new RuntimeException ( e ) ;
}
}
long timestampToUse = lastTimestamp ;
lastTimestamp = System . currentTimeMillis ( ) ;
long start = System . currentTimeMillis ( ) ;
ses = repository . login ( credentials ) ;
final String [ ] types = { "nthl:workspaceLeafItem" , "nthl:workspaceItem" } ;
private void update ( Session ses , Map < String , Long > tempCachedMap ) {
try {
long timestampToUse = lastTimestamp ;
lastTimestamp = System . currentTimeMillis ( ) ;
EventJournal journalChanged = ses . getWorkspace ( ) . getObservationManager ( ) . getEventJournal ( Event . PROPERTY_CHANGED ^ Event . NODE_REMOVED ^ Event . NODE_MOVED ^ Event . NODE_ADDED , vreFolder . getPath ( ) , true , null , types ) ;
journalChanged . skipTo ( timestampToUse ) ;
long start = System . currentTimeMillis ( ) ;
final String [ ] types = { "nthl:workspaceLeafItem" , "nthl:workspaceItem" } ;
logger . trace ( "getting the journal took {}" , System . currentTimeMillis ( ) - start ) ;
EventJournal journalChanged = ses . getWorkspace ( ) . getObservationManager ( ) . getEventJournal ( Event . PROPERTY_CHANGED ^ Event . NODE_REMOVED ^ Event . NODE_MOVED ^ Event . NODE_ADDED , vreFolder . getPath ( ) , true , null , types ) ;
journalChanged . skipTo ( timestampToUse ) ;
int events = 0 ;
logger . debug ( "getting the journal took {}" , System . currentTimeMillis ( ) - start ) ;
while ( journalChanged . hasNext ( ) ) {
events + + ;
Event event = journalChanged . nextEvent ( ) ;
int events = 0 ;
//TODO: manage hidden nodes
while ( journalChanged . hasNext ( ) ) {
events + + ;
Event event = journalChanged . nextEvent ( ) ;
try {
switch ( event . getType ( ) ) {
case Event . NODE_ADDED :
if ( ses . nodeExists ( event . getPath ( ) ) ) {
Node nodeAdded = ses . getNode ( event . getPath ( ) ) ;
if ( nodeAdded . isNodeType ( "nthl:workspaceLeafItem" ) ) {
logger . trace ( "node added event received with name {}" , nodeAdded . getName ( ) ) ;
Item item = node2Item . getItem ( nodeAdded , Arrays . asList ( NodeConstants . ACCOUNTING_NAME ) ) ;
insertItemInTheRightPlace ( item ) ;
logger . debug ( "node added event received with name {}" , nodeAdded . getName ( ) ) ;
Item item = node2Item . getItem ( nodeAdded , Excludes . ALL ) ;
if ( tempCachedMap . get ( event . getIdentifier ( ) ) ! = null )
tempCachedMap . remove ( event . getIdentifier ( ) ) ;
insertItemInTheRightPlace ( item , tempCachedMap ) ;
}
}
break ;
@ -131,66 +194,64 @@ public class VREQueryRetriever implements Callable<List<Item>> {
if ( ses . propertyExists ( event . getPath ( ) ) ) {
Property property = ses . getProperty ( event . getPath ( ) ) ;
if ( property . getName ( ) . equalsIgnoreCase ( "jcr:lastModified" ) ) {
logger . trace ( "event property changed on {} with value {} and parent {}" , property . getName ( ) , property . getValue ( ) . getString ( ) , property . getParent ( ) . getPath ( ) ) ;
logger . debug ( "event property changed on {} with value {} and parent {}" , property . getName ( ) , property . getValue ( ) . getString ( ) , property . getParent ( ) . getPath ( ) ) ;
String identifier = property . getParent ( ) . getIdentifier ( ) ;
cachedList. removeIf ( i - > i . getId ( ) . equals ( identifier ) ) ;
Item item = node2Item . getItem ( property . getParent ( ) , Excludes . EXCLUDE_ACCOUNTING ) ;
insertItemInTheRightPlace ( item ;
tempCachedMap. remove ( identifier ) ;
Item item = node2Item . getItem ( property . getParent ( ) , Excludes . ALL ) ;
insertItemInTheRightPlace ( item , tempCachedMap );
}
}
break ;
case Event . NODE_REMOVED :
logger . trace ( "node removed event received with type {}" , event . getIdentifier ( ) ) ;
cachedList . removeIf ( i - > {
try {
return i . getId ( ) . equals ( event . getIdentifier ( ) ) & & i . getLastModificationTime ( ) . getTime ( ) . getTime ( ) < event . getDate ( ) ;
} catch ( RepositoryException e ) {
return false ;
}
} ) ;
if ( tempCachedMap . get ( event . getIdentifier ( ) ) ! = null & &
tempCachedMap . get ( event . getIdentifier ( ) ) < event . getDate ( ) )
tempCachedMap . remove ( event . getIdentifier ( ) ) ;
break ;
case Event . NODE_MOVED :
Node nodeMoved = ses . getNode ( event . getPath ( ) ) ;
logger . trace ( "node moved event received with type {}" , nodeMoved . getPrimaryNodeType ( ) ) ;
if ( nodeMoved . isNodeType ( "nthl:workspaceLeafItem" ) ) {
logger . trace ( "event node moved on {} with path {}" , nodeMoved . getName ( ) , nodeMoved . getPath ( ) ) ;
logger . debug ( "event node moved on {} with path {}" , nodeMoved . getName ( ) , nodeMoved . getPath ( ) ) ;
String identifier = nodeMoved . getIdentifier ( ) ;
cachedList . removeIf ( i - > i . getId ( ) . equals ( identifier ) & & ! i . getPath ( ) . startsWith ( vreFolder . getPath ( ) ) ) ;
String nodePath = ses . getNode ( identifier ) . getPath ( ) ;
if ( tempCachedMap . get ( event . getIdentifier ( ) ) ! = null & &
! nodePath . startsWith ( vreFolder . getPath ( ) ) )
tempCachedMap . remove ( event . getIdentifier ( ) ) ;
}
break ;
default :
throw new Exception ( "error in event handling" ) ;
break;
}
} catch ( Exception e ) {
logger . warn ( "error handling event {}" , event . getType ( ) , e ) ;
}
if ( cachedList . size ( ) > CACHE_DIMENSION )
cachedList . subList ( 51 , cachedList . size ( ) ) . clear ( ) ;
logger . trace ( "retrieving event took {} with {} events" , System . currentTimeMillis ( ) - start , events ) ;
if ( cachedList . size ( ) < = 10 ) return cachedList ;
else return cachedList . subList ( 0 , 10 ) ;
} catch ( Exception e ) {
logger . error ( "error getting events for vre {}" , vreFolder . getTitle ( ) , e ) ;
throw new RuntimeException ( e ) ;
} finally {
if ( ses ! = null )
ses . logout ( ) ;
}
logger . trace ( "retrieving event took {} with {} events" , System . currentTimeMillis ( ) - start , events ) ;
} catch ( Exception e ) {
logger . error ( "error getting events for vre {}" , vreFolder . getTitle ( ) , e ) ;
throw new RuntimeException ( e ) ;
}
}
private void insertItemInTheRightPlace ( Item item ) {
Iterator < Item > it = cachedList . iterator ( ) ;
int index = 0 ;
while ( it . hasNext ( ) ) {
Item inListItem = it . next ( ) ;
if ( item . getLastModificationTime ( ) . getTime ( ) . getTime ( ) > = inListItem . getLastModificationTime ( ) . getTime ( ) . getTime ( ) ) break ;
index + + ;
private void visitChildren ( Node node , Map < String , Long > tempCachedMap ) throws Exception {
NodeIterator nodeIt = node . getNodes ( ) ;
while ( nodeIt . hasNext ( ) ) {
Node child = nodeIt . nextNode ( ) ;
Item item = node2Item . getItem ( child , Excludes . ALL ) ;
if ( item = = null | | item . isHidden ( ) ) continue ;
if ( item instanceof FolderItem )
visitChildren ( child , tempCachedMap ) ;
else if ( item instanceof AbstractFileItem )
insertItemInTheRightPlace ( item , tempCachedMap ) ;
}
if ( index < CACHE_DIMENSION )
cachedList . add ( index , item ) ;
}
/ * @Override
lucio.lelii
3 years ago