Skip to content

Commit

Permalink
cells: Minor refactoring of CellPath
Browse files Browse the repository at this point in the history 
Target: trunk
Require-notes: no
Require-book: no
Acked-by: Paul Millar <paul.millar@desy.de>
Patch: https://rb.dcache.org/r/8046/
  • Loading branch information
@gbehrmann
gbehrmann committed Apr 1, 2015
1 parent 05e9bd9 commit f5da9d4
Show file tree
Hide file tree
Showing 6 changed files with 48 additions and 61 deletions.
84 changes: 36 additions & 48 deletions modules/cells/src/main/java/dmg/cells/nucleus/CellPath.java
View file
Expand Up @@ -4,9 +4,12 @@

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.StringTokenizer;
import java.util.stream.Stream;

import static com.google.common.base.Preconditions.checkArgument;
import static java.util.stream.Collectors.toList;

/**
*
* The CellPath is an abstraction of the path a CellMessage is
Expand Down Expand Up @@ -42,80 +45,61 @@ private CellPath(int position, List<CellAddressCore> list)
_list = list;
}

protected CellPath(CellPath addr)
{
this(addr._position, Lists.newArrayList(addr._list));
}

protected CellPath()
{
this(-1, new ArrayList<CellAddressCore>());
this(-1, new ArrayList<>());
}

public CellPath(String path)
{
this();
add(path);
this(0, streamOfPath(path).collect(toList()));
}

public CellPath(CellAddressCore address)
{
this();
add(address);
this(0, Lists.newArrayList(address));
}

public CellPath(String cellName, String domainName)
{
this(new CellAddressCore(cellName, domainName));
}

public int hops()
public synchronized int hops()
{
return _list.size();
}

public synchronized void add(CellAddressCore core)
{
_list.add(core) ;
if( _position < 0 ) {
_position = 0;
}
}
public synchronized void add(CellAddressCore core)
{
_list.add(core) ;
if (_position < 0) {
_position = 0;
}
}

public synchronized void add( CellPath addr ){
_list.addAll( addr._list ) ;
if( _position < 0 ) {
_position = 0;
}
}
public synchronized void add(CellPath addr)
{
_list.addAll(addr._list) ;
if (_position < 0) {
_position = 0;
}
}

public void add( String cell , String domain ){
add(new CellAddressCore(cell, domain)) ;
}
/**
/**
* Adds a cell path &lt;path&gt; to the end of the current path.
*
* @param path The added cell travel path.
*/
public synchronized void add( String path ){
StringTokenizer st = new StringTokenizer( path ,":" ) ;
while (st.hasMoreTokens()) {
add(new CellAddressCore(st.nextToken()));
}
public synchronized void add(String path)
{
streamOfPath(path).forEachOrdered(this::add);
}
/**
* Creates a CellAddress with a single cell as initial destination.
* The cell is represented by its name and the name of its domain.
*
* @param cellName The name of the initial destination cell.
* @param domainName The name of the initial destination cells domain.
*/

@Override
public synchronized Object clone(){
CellPath addr = new CellPath() ;
addr._list.addAll(_list);
addr._position = _position ;
return addr ;
public synchronized Object clone()
{
return new CellPath(_position, Lists.newArrayList(_list));
}

/**
Expand Down Expand Up @@ -262,5 +246,9 @@ public synchronized int hashCode(){
return (_list.hashCode() * 17) ^ _position;
}


private static Stream<CellAddressCore> streamOfPath(String path)
{
checkArgument(!path.isEmpty());
return Stream.of(path.split(":")).map(CellAddressCore::new);
}
}
3 changes: 1 addition & 2 deletions modules/cells/src/main/java/dmg/cells/nucleus/EventLogger.java
View file
Expand Up @@ -80,8 +80,7 @@ public static void sendBegin(CellNucleus nucleus, CellMessage envelope, String m
*/
CellPath source = envelope.getSourcePath();
if (source.hops() == 0) {
source = new CellPath();
source.add(nucleus.getThisAddress());
source = new CellPath(nucleus.getThisAddress());
}
NetLoggerBuilder log = new NetLoggerBuilder(SEND_BEGIN);
log.add("uoid", envelope.getUOID());
Expand Down
14 changes: 7 additions & 7 deletions modules/dcache/src/test/java/org/dcache/tests/poolmanager/HsmRestoreTest.java
View file
Expand Up @@ -235,7 +235,7 @@ public void testRestoreNoLocationsOnePoolCantStage() throws Exception {
poolCostInfo.setQueueSizes(0, 10, 0, 0, 10, 0);
poolCostInfo.addExtendedMoverQueueSizes(IoQueueManager.DEFAULT_QUEUE, 0, 10, 0, 0, 0);

CellMessage envelope = new CellMessage(new CellPath(""), null);
CellMessage envelope = new CellMessage(new CellPath("PoolManager"), null);
envelope.addSourceAddress(new CellAddressCore(pool));
PoolManagerPoolUpMessage poolUpMessage = new PoolManagerPoolUpMessage(pool, serialId, poolMode, poolCostInfo);

Expand All @@ -258,7 +258,7 @@ public void testRestoreNoLocationsOnePoolCantStage() throws Exception {

PoolMgrSelectReadPoolMsg selectReadPool = new PoolMgrSelectReadPoolMsg(attributes, _protocolInfo, null);
CellMessage cellMessage = new CellMessage( new CellPath("PoolManager"), selectReadPool);
cellMessage.getSourcePath().add("door", "local");
cellMessage.getSourcePath().add(new CellAddressCore("door", "local")) ;

_rc.messageArrived(cellMessage, selectReadPool);

Expand Down Expand Up @@ -335,7 +335,7 @@ public void testRestoreNoLocationsSinglePool() throws Exception {
poolCostInfo.setQueueSizes(0, 10, 0, 0, 10, 0);
poolCostInfo.addExtendedMoverQueueSizes(IoQueueManager.DEFAULT_QUEUE, 0, 10, 0, 0, 0);

CellMessage envelope = new CellMessage(new CellPath(""), null);
CellMessage envelope = new CellMessage(new CellPath("PoolManager"), null);
envelope.addSourceAddress(new CellAddressCore(pool));
PoolManagerPoolUpMessage poolUpMessage = new PoolManagerPoolUpMessage(pool, serialId, poolMode, poolCostInfo);

Expand All @@ -355,7 +355,7 @@ public void testRestoreNoLocationsSinglePool() throws Exception {

PoolMgrSelectReadPoolMsg selectReadPool = new PoolMgrSelectReadPoolMsg(attributes, _protocolInfo, null);
CellMessage cellMessage = new CellMessage( new CellPath("PoolManager"), selectReadPool);
cellMessage.getSourcePath().add("door", "local");
cellMessage.getSourcePath().add(new CellAddressCore("door", "local")) ;

_rc.messageArrived(cellMessage, selectReadPool);

Expand Down Expand Up @@ -434,7 +434,7 @@ public void testRestoreNoLocationsAllPoolsCantStage() throws Exception {
poolCostInfo.setQueueSizes(0, 10, 0, 0, 10, 0);
poolCostInfo.addExtendedMoverQueueSizes(IoQueueManager.DEFAULT_QUEUE, 0, 10, 0, 0, 0);

CellMessage envelope = new CellMessage(new CellPath(""), null);
CellMessage envelope = new CellMessage(new CellPath("PoolManager"), null);
envelope.addSourceAddress(new CellAddressCore(pool));
PoolManagerPoolUpMessage poolUpMessage = new PoolManagerPoolUpMessage(pool, serialId, poolMode, poolCostInfo);

Expand All @@ -457,7 +457,7 @@ public void testRestoreNoLocationsAllPoolsCantStage() throws Exception {

PoolMgrSelectReadPoolMsg selectReadPool = new PoolMgrSelectReadPoolMsg(attributes, _protocolInfo, null);
CellMessage cellMessage = new CellMessage( new CellPath("PoolManager"), selectReadPool);
cellMessage.getSourcePath().add("door", "local");
cellMessage.getSourcePath().add(new CellAddressCore("door", "local")) ;

_rc.messageArrived(cellMessage, selectReadPool);

Expand All @@ -476,7 +476,7 @@ public void testRestoreNoLocationsAllPoolsCantStage() throws Exception {
// resubmit request
PoolMgrSelectReadPoolMsg selectReadPool2 = new PoolMgrSelectReadPoolMsg(attributes, _protocolInfo, selectReadPool.getContext());
CellMessage cellMessage2 = new CellMessage( new CellPath("PoolManager"), selectReadPool2);
cellMessage2.getSourcePath().add("door", "local");
cellMessage2.getSourcePath().add(new CellAddressCore("door", "local")) ;
_rc.messageArrived(cellMessage2, selectReadPool2);

// second pool replies with an error
Expand Down
2 changes: 1 addition & 1 deletion modules/dcache/src/test/java/org/dcache/tests/poolmanager/PoolCostInfoHelper.java
View file
Expand Up @@ -24,7 +24,7 @@ public static void setCost(CostModuleV1 cm, String pool,long total , long free ,

poolCost.setSpaceUsage(total, free, precious, removable);

CellMessage envelope = new CellMessage(new CellPath(""), null);
CellMessage envelope = new CellMessage(new CellPath("PoolManager"), null);
envelope.addSourceAddress(new CellAddressCore(pool));
PoolManagerPoolUpMessage poolUpMessage = new PoolManagerPoolUpMessage(pool,
serialId, poolMode, poolCost);
Expand Down
4 changes: 2 additions & 2 deletions modules/dcache/src/test/java/org/dcache/tests/poolmanager/PoolMonitorTest.java
View file
Expand Up @@ -87,9 +87,9 @@ public void testFromStore() throws Exception {
PoolManagerPoolUpMessage pool2UpMessage = new PoolManagerPoolUpMessage("pool2",
serialId, poolMode, poolCost2);

CellMessage envelope1 = new CellMessage(new CellPath(""), null);
CellMessage envelope1 = new CellMessage(new CellPath("PoolManager"), null);
envelope1.addSourceAddress(new CellAddressCore("pool1"));
CellMessage envelope2 = new CellMessage(new CellPath(""), null);
CellMessage envelope2 = new CellMessage(new CellPath("PoolManager"), null);
envelope2.addSourceAddress(new CellAddressCore("pool2"));

_costModule.messageArrived(envelope1, pool1UpMessage);
Expand Down
2 changes: 1 addition & 1 deletion modules/dcache/src/test/java/org/dcache/tests/util/CellMessageDispatcherTest.java
View file
Expand Up @@ -154,7 +154,7 @@ private Object deliver(Serializable msg, int result1, int result2)
{
try {
return
dispatcher.call(new CellMessage(new CellPath(""), msg));
dispatcher.call(new CellMessage(new CellPath("test"), msg));
} finally {
assertEquals(listener1.delivered, result1);
assertEquals(listener2.delivered, result2);
Expand Down

0 comments on commit f5da9d4

Please sign in to comment.