Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • scs/ipaaca
  • ramin.yaghoubzadeh/ipaaca
2 results
Show changes
Hide whitespace changes
Inline Side-by-side
Showing
with 1019 additions and 135 deletions
ipaacalib/java/src/ipaaca/BufferConfiguration.java
View file @ 460357ca
......@@ -33,6 +33,7 @@
package ipaaca;
import java.util.HashSet;
import java.util.Set;
......@@ -50,6 +51,7 @@ class BufferConfiguration {
this._owningComponentName = owningComponentName;
this._channel = "default";
this._resendActive = false;
this._category_interests = new HashSet<String>();
}
public String getOwningComponentName() {
......
ipaacalib/java/src/ipaaca/BufferConfigurationBuilder.java
View file @ 460357ca
......@@ -33,7 +33,7 @@
package ipaaca;
class BufferConfigurationBuilder extends BufferConfiguration {
public class BufferConfigurationBuilder extends BufferConfiguration {
public BufferConfigurationBuilder(String owningComponentName) {
super(owningComponentName);
......
ipaacalib/java/src/ipaaca/IUConverter.java
View file @ 460357ca
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* CITEC, Bielefeld University
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
......@@ -21,7 +21,7 @@
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
......@@ -55,7 +55,7 @@ import com.google.protobuf.InvalidProtocolBufferException;
/**
* Serializes AbstractIUs into protocolbuffer IUs and vice versa.
* @author hvanwelbergen
*
*
*/
public class IUConverter implements Converter<ByteBuffer>
{
......@@ -88,18 +88,15 @@ public class IUConverter implements Converter<ByteBuffer>
links.add(LinkSet.newBuilder().setType(entry.getKey()).addAllTargets(entry.getValue()).build());
}
IU.AccessMode accessMode = IU.AccessMode.PUSH;
if(iua instanceof RemoteMessageIU || iua instanceof LocalMessageIU)
{
accessMode = IU.AccessMode.MESSAGE;
}
IU iu = IU.newBuilder().setUid(iua.getUid()).setRevision(iua.getRevision()).setCategory(iua.getCategory())
IU.AccessMode accessMode = iua.getAccessMode();
IU iu = IU.newBuilder().setUid(iua.getUid()).setRevision((int) iua.getRevision()).setCategory(iua.getCategory())
.setOwnerName(iua.getOwnerName()).setCommitted(iua.isCommitted()).setAccessMode(accessMode)
.setReadOnly(iua.isReadOnly()).setPayloadType("STR").addAllPayload(payloadItems).addAllLinks(links).build();
return new WireContents<ByteBuffer>(ByteBuffer.wrap(iu.toByteArray()), "ipaaca-iu");
String wireFormat = (accessMode == IU.AccessMode.MESSAGE) ? "ipaaca-messageiu" : "ipaaca-iu";
return new WireContents<ByteBuffer>(ByteBuffer.wrap(iu.toByteArray()), wireFormat);
}
@Override
public UserData<?> deserialize(String wireSchema, ByteBuffer buffer) throws ConversionException
{
......@@ -107,6 +104,12 @@ public class IUConverter implements Converter<ByteBuffer>
try
{
iu = IU.newBuilder().mergeFrom(buffer.array()).build();
// If there are rsb buffer read-only issues in some build, use this code instead of the above line:
//int size = buffer.capacity();
//byte[] array = new byte[size];
//buffer.get(array, 0, size);
//iu = IU.newBuilder().mergeFrom(array).build();
}
catch (InvalidProtocolBufferException e)
{
......
ipaacalib/cpp/include/ipaaca/ipaaca-json.h → ipaacalib/java/src/ipaaca/IURetractedException.java
View file @ 460357ca
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* (formerly the Sociable Agents Group)
* CITEC, Bielefeld University
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
......@@ -22,7 +21,7 @@
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
......@@ -31,20 +30,25 @@
* Excellence Initiative.
*/
#ifndef __ipaaca_json_H__
#define __ipaaca_json_H__
#include "rapidjson/document.h"
#include "rapidjson/prettywriter.h"
#include "rapidjson/filestream.h"
#include <cstdio>
// Notes:
// - From http://stackoverflow.com/questions/10426924/json-root-element
// Actually there are two different JSON specifications. RFC 4627 requires a JSON text to be
// an object or an array. ECMA-262, 5th edition, section 15.12 does not impose this restriction.
package ipaaca;
/**
* Error indicating that an IU is immutable because it has been retracted.
*
*/
public class IURetractedException extends RuntimeException
{
private static final long serialVersionUID = 1L;
private final AbstractIU iu;
#endif // __IPAACA_JSON_H__
public AbstractIU getIU()
{
return iu;
}
public IURetractedException(AbstractIU iu)
{
super("Writing to IU " + iu.getUid() + " failed -- it has been retracted.");
this.iu = iu;
}
}
ipaacalib/java/src/ipaaca/Initializer.java
View file @ 460357ca
......@@ -3,7 +3,7 @@
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
......@@ -32,9 +32,12 @@
package ipaaca;
import java.nio.ByteBuffer;
import ipaaca.protobuf.Ipaaca.IUCommission;
import ipaaca.protobuf.Ipaaca.IUResendRequest;
import ipaaca.protobuf.Ipaaca.IURetraction;
import rsb.converter.ConverterSignature;
import rsb.converter.ConverterRepository;
import rsb.converter.DefaultConverterRepository;
import rsb.converter.ProtocolBufferConverter;
......@@ -43,34 +46,74 @@ import rsb.converter.ProtocolBufferConverter;
* @author hvanwelbergen
*
*/
public final class Initializer
{
private Initializer()
{
}
public final class Initializer {
private Initializer() {}
private static volatile boolean initialized = false;
public synchronized static void initializeIpaacaRsb()
{
if(initialized)return;
DefaultConverterRepository.getDefaultConverterRepository().addConverter(new IntConverter());
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new ProtocolBufferConverter<IUCommission>(IUCommission.getDefaultInstance()));
// dlw
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new ProtocolBufferConverter<IUResendRequest>(IUResendRequest.getDefaultInstance()));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new IUConverter(new ConverterSignature("ipaaca-iu", RemotePushIU.class)));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new IUConverter(new ConverterSignature("ipaaca-localiu", LocalIU.class)));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new IUConverter(new ConverterSignature("ipaaca-messageiu", RemoteMessageIU.class)));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(
new IUConverter(new ConverterSignature("ipaaca-localmessageiu", LocalMessageIU.class)));
public synchronized static void initializeIpaacaRsb() {
if (initialized)
return;
ConverterRepository<ByteBuffer> dcr =
DefaultConverterRepository.getDefaultConverterRepository();
// for IU revision numbers
dcr.addConverter(
new IntConverter());
// IU commit messages
dcr.addConverter(
new ProtocolBufferConverter<IUCommission>(
IUCommission.getDefaultInstance()));
// IU commit messages
dcr.addConverter(
new ProtocolBufferConverter<IURetraction>(
IURetraction.getDefaultInstance()));
// IU resend request messages
dcr.addConverter(
new ProtocolBufferConverter<IUResendRequest>(
IUResendRequest.getDefaultInstance()));
// IUs
dcr.addConverter(
new IUConverter(
new ConverterSignature(
"ipaaca-iu",
RemotePushIU.class)));
// Local IUs
dcr.addConverter(
new IUConverter(
new ConverterSignature(
"ipaaca-localiu",
LocalIU.class)));
// Messages
dcr.addConverter(
new IUConverter(
new ConverterSignature(
"ipaaca-messageiu",
RemoteMessageIU.class)));
// LocalMessages
dcr.addConverter(
new IUConverter(
new ConverterSignature(
"ipaaca-localmessageiu",
LocalMessageIU.class)));
DefaultConverterRepository.getDefaultConverterRepository().addConverter(new PayloadConverter());
DefaultConverterRepository.getDefaultConverterRepository().addConverter(new LinkUpdateConverter());
// Payloads
dcr.addConverter(
new PayloadConverter());
// LinkUpdates
dcr.addConverter(
new LinkUpdateConverter());
initialized = true;
}
}
ipaacalib/java/src/ipaaca/InputBuffer.java
View file @ 460357ca
......@@ -33,6 +33,7 @@
package ipaaca;
import ipaaca.protobuf.Ipaaca.IUCommission;
import ipaaca.protobuf.Ipaaca.IURetraction;
import ipaaca.protobuf.Ipaaca.IUResendRequest;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
......@@ -441,6 +442,22 @@ public class InputBuffer extends Buffer
iu.setRevision(iuc.getRevision());
callIuEventHandlers(iuc.getUid(), false, IUEventType.COMMITTED, iu.getCategory());
}
if (event.getData() instanceof IURetraction)
{
IURetraction iuc = (IURetraction) event.getData();
logger.debug("handleIUEvents invoked with an IURetraction: {}", iuc);
logger.debug("{}", this.getUniqueName());
if (!iuStore.containsKey(iuc.getUid()))
{
logger.warn("Update message for IU which we did not fully receive before.");
}
RemotePushIU iu = this.iuStore.get(iuc.getUid());
if (iu != null) {
iu.applyRetraction();
callIuEventHandlers(iuc.getUid(), false, IUEventType.RETRACTED, iu.getCategory());
}
}
}
}
......@@ -466,10 +483,10 @@ public class InputBuffer extends Buffer
rServer = getRemoteServer(writerName);
if ((rServer != null)&&(uid != null)) {
IUResendRequest iurr = IUResendRequest.newBuilder().setUid(uid).setHiddenScopeName(hiddenScopeName).build();
int rRevision = 0;
long rRevision = 0;
try
{
rRevision = (Integer) rServer.call("resendRequest", iurr);
rRevision = (Long) rServer.call("resendRequest", iurr);
}
catch (RSBException e)
{
......@@ -483,6 +500,10 @@ public class InputBuffer extends Buffer
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (rRevision == 0)
{
//throw new IUResendFailedException(aiu); // TODO
......@@ -508,6 +529,14 @@ public class InputBuffer extends Buffer
}
}
public void addCategoryInterest(String... categories)
{
for(String cat:categories)
{
createCategoryListenerIfNeeded(cat);
}
}
public Collection<RemotePushIU> getIUs()
{
return iuStore.values();
......
ipaacalib/java/src/ipaaca/LocalIU.java
View file @ 460357ca
......@@ -32,6 +32,7 @@
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate.Builder;
......@@ -50,6 +51,11 @@ import com.google.common.collect.SetMultimap;
public class LocalIU extends AbstractIU
{
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.PUSH;
}
private OutputBuffer outputBuffer;
......@@ -105,6 +111,10 @@ public class LocalIU extends AbstractIU
synchronized (revisionLock)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
if (committed)
{
throw new IUCommittedException(this);
......@@ -121,6 +131,22 @@ public class LocalIU extends AbstractIU
}
}
private void internalRetract()
{
synchronized (revisionLock)
{
if (isRetracted())
return;
increaseRevisionNumber();
retracted = true;
if (outputBuffer != null)
{
outputBuffer.sendIURetraction(this);
}
}
}
private void increaseRevisionNumber()
{
revision++;
......@@ -148,6 +174,10 @@ public class LocalIU extends AbstractIU
@Override
void modifyLinks(boolean isDelta, SetMultimap<String, String> linksToAdd, SetMultimap<String, String> linksToRemove, String writerName)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isCommitted())
{
throw new IUCommittedException(this);
......@@ -181,7 +211,7 @@ public class LocalIU extends AbstractIU
removeSet.add(LinkSet.newBuilder().setType(entry.getKey()).addAllTargets(entry.getValue()).build());
}
outputBuffer.sendIULinkUpdate(this,
IULinkUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setWriterName(wName).setIsDelta(isDelta)
IULinkUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setWriterName(wName).setIsDelta(isDelta)
.addAllNewLinks(addSet).addAllLinksToRemove(removeSet).build());
}
}
......@@ -226,13 +256,17 @@ public class LocalIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
increaseRevisionNumber();
if (isPublished())
{
// send update to remote holders
PayloadItem newItem = PayloadItem.newBuilder().setKey(key).setValue(value).setType("STR")
.build();
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(true)
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(true)
.setWriterName(writer == null ? getOwnerName() : writer).addNewItems(newItem).build();
getOutputBuffer().sendIUPayloadUpdate(this, update);
}
......@@ -249,10 +283,14 @@ public class LocalIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
increaseRevisionNumber();
if (isPublished())
{
Builder builder = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(true)
Builder builder = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(true)
.setWriterName(writer == null ? getOwnerName() : writer);
for (Map.Entry<? extends String, ? extends String> item : newItems.entrySet())
{
......@@ -270,15 +308,29 @@ public class LocalIU extends AbstractIU
@Override
public void commit()
{
if (isRetracted())
{
throw new IURetractedException(this);
}
internalCommit(null);
}
@Override
public void commit(String writerName)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
internalCommit(writerName);
}
@Override
public void retract()
{
internalRetract();
}
@Override
void removeFromPayload(Object key, String writer)
{
......@@ -288,11 +340,15 @@ public class LocalIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
increaseRevisionNumber();
if (isPublished())
{
// send update to remote holders
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(true)
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(true)
.setWriterName(writer == null ? getOwnerName() : writer).addKeysToRemove((String) key).build();
getOutputBuffer().sendIUPayloadUpdate(this, update);
}
......@@ -305,7 +361,7 @@ public class LocalIU extends AbstractIU
{
if (isPublished())
{
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(false)
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(false)
.setWriterName(writerName == null ? getOwnerName() : writerName).addAllNewItems(newPayload).build();
getOutputBuffer().sendIUPayloadUpdate(this, update);
}
......
ipaacalib/java/src/ipaaca/LocalMessageIU.java
View file @ 460357ca
......@@ -3,7 +3,7 @@
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
......@@ -32,6 +32,7 @@
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
/**
* Local IU of Message sub-type. Can be handled like a normal IU, but on the remote side it is only existent during the handler calls.
......@@ -50,4 +51,9 @@ public class LocalMessageIU extends LocalIU
super(category);
}
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.MESSAGE;
}
}
ipaacalib/java/src/ipaaca/OutputBuffer.java
View file @ 460357ca
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
......@@ -34,6 +35,7 @@ package ipaaca;
import ipaaca.protobuf.Ipaaca;
import ipaaca.protobuf.Ipaaca.IUCommission;
import ipaaca.protobuf.Ipaaca.IURetraction;
import ipaaca.protobuf.Ipaaca.IUResendRequest;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
......@@ -53,7 +55,9 @@ import rsb.Informer;
import rsb.InitializeException;
import rsb.RSBException;
import rsb.patterns.DataCallback;
import rsb.patterns.EventCallback;
import rsb.patterns.LocalServer;
import rsb.Event;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.SetMultimap;
......@@ -129,48 +133,58 @@ public class OutputBuffer extends Buffer
this.channel = ipaaca_channel;
}
private final class RemoteUpdatePayload extends DataCallback<Integer, IUPayloadUpdate>
private final class RemoteUpdatePayload extends EventCallback //DataCallback<Long, IUPayloadUpdate>
{
@Override
public Integer invoke(IUPayloadUpdate data) throws Throwable
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteUpdate");
return remoteUpdatePayload(data);
long result = remoteUpdatePayload((IUPayloadUpdate) request.getData());
//System.out.println("remoteUpdatePayload yielded revision "+result);
return new Event(Long.class, new Long(result));
}
}
/*private final class RemoteUpdatePayload extends DataCallback<Long, IUPayloadUpdate>
{
@Override
public Long invoke(IUPayloadUpdate data) throws Throwable
{
logger.debug("remoteUpdate");
return remoteUpdatePayload(data);
}
}*/
private final class RemoteUpdateLinks extends DataCallback<Integer, IULinkUpdate>
private final class RemoteUpdateLinks extends EventCallback // DataCallback<Long, IULinkUpdate>
{
@Override
public Integer invoke(IULinkUpdate data) throws Throwable
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteUpdateLinks");
return remoteUpdateLinks(data);
return new Event(Long.class, new Long(remoteUpdateLinks((IULinkUpdate) request.getData())));
}
}
private final class RemoteCommit extends DataCallback<Integer, IUCommission>
private final class RemoteCommit extends EventCallback //DataCallback<Long, IUCommission>
{
@Override
public Integer invoke(IUCommission data) throws Throwable
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteCommit");
return remoteCommit(data);
return new Event(Long.class, new Long(remoteCommit((IUCommission) request.getData())));
}
}
private final class RemoteResendRequest extends DataCallback<Integer, IUResendRequest>
private final class RemoteResendRequest extends EventCallback //DataCallback<Long, IUResendRequest>
{
@Override
public Integer invoke(IUResendRequest data) throws Throwable
public Event invoke(final Event request) //throws Throwable
{
logger.debug("remoteResendRequest");
return remoteResendRequest(data);
return new Event(Long.class, new Long(remoteResendRequest((IUResendRequest) request.getData())));
}
}
// def _remote_update_payload(self, update):
......@@ -197,7 +211,7 @@ public class OutputBuffer extends Buffer
* Apply a remotely requested update to one of the stored IUs.
* @return 0 if not updated, IU version number otherwise
*/
int remoteUpdatePayload(IUPayloadUpdate update)
long remoteUpdatePayload(IUPayloadUpdate update)
{
if (!iuStore.containsKey(update.getUid()))
{
......@@ -242,7 +256,7 @@ public class OutputBuffer extends Buffer
* Apply a remotely requested update to one of the stored IUs.
* @return 0 if not updated, IU version number otherwise
*/
int remoteUpdateLinks(IULinkUpdate update)
long remoteUpdateLinks(IULinkUpdate update)
{
if (!iuStore.containsKey(update.getUid()))
{
......@@ -305,7 +319,7 @@ public class OutputBuffer extends Buffer
/**
* Apply a remotely requested commit to one of the stored IUs.
*/
private int remoteCommit(IUCommission iuc)
private long remoteCommit(IUCommission iuc)
{
if (!iuStore.containsKey(iuc.getUid()))
{
......@@ -334,7 +348,7 @@ public class OutputBuffer extends Buffer
/*
* Resend an requested iu over the specific hidden channel. (dlw) TODO
*/
private int remoteResendRequest(IUResendRequest iu_resend_request_pack)
private long remoteResendRequest(IUResendRequest iu_resend_request_pack)
{
if (!iuStore.containsKey(iu_resend_request_pack.getUid()))
{
......@@ -347,7 +361,7 @@ public class OutputBuffer extends Buffer
Informer<Object> informer = getInformer(iu_resend_request_pack.getHiddenScopeName());
try
{
informer.send(iu);
informer.publish(iu);
}
catch (RSBException e)
{
......@@ -402,6 +416,10 @@ public class OutputBuffer extends Buffer
{
throw new RuntimeException(e);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
return informer;
}
......@@ -439,7 +457,7 @@ public class OutputBuffer extends Buffer
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(iu);
informer.publish(iu);
}
catch (RSBException e)
{
......@@ -473,12 +491,26 @@ public class OutputBuffer extends Buffer
*/
protected void sendIUCommission(AbstractIU iu, String writerName)
{
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(iu.getUid()).setRevision(iu.getRevision())
.setWriterName(iu.getOwnerName() != null ? iu.getOwnerName() : writerName).build();
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(iu.getUid()).setRevision((int) iu.getRevision())
.setWriterName(writerName == null ? iu.getOwnerName() : writerName).build();
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.publish(iuc);
}
catch (RSBException e)
{
throw new RuntimeException(e);
}
}
protected void sendIURetraction(AbstractIU iu)
{
IURetraction iuc = Ipaaca.IURetraction.newBuilder().setUid(iu.getUid()).setRevision((int) iu.getRevision()).build();
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(iuc);
informer.publish(iuc);
}
catch (RSBException e)
{
......@@ -518,7 +550,7 @@ public class OutputBuffer extends Buffer
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(update);
informer.publish(update);
}
catch (RSBException e)
{
......@@ -531,7 +563,7 @@ public class OutputBuffer extends Buffer
Informer<Object> informer = getInformer(iu.getCategory());
try
{
informer.send(update);
informer.publish(update);
}
catch (RSBException e)
{
......
ipaacalib/java/src/ipaaca/Payload.java
View file @ 460357ca
......@@ -34,7 +34,12 @@ package ipaaca;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import org.apache.commons.lang.StringEscapeUtils;
import com.google.common.collect.ImmutableSet;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
......@@ -47,7 +52,7 @@ import java.util.Set;
*/
public class Payload implements Map<String, String>
{
private Map<String, String> map = new HashMap<String, String>();
private Map<String, String> map = Collections.synchronizedMap(new HashMap<String, String>());
private final AbstractIU iu;
public Payload(AbstractIU iu)
......@@ -80,31 +85,38 @@ public class Payload implements Map<String, String>
public void set(Map<String, String> newPayload, String writerName)
{
iu.setPayload(newPayload, writerName);
map.clear();
map.putAll(newPayload);
synchronized(map)
{
map.clear();
map.putAll(newPayload);
}
}
public void set(List<PayloadItem> newPayload, String writerName)
{
iu.handlePayloadSetting(newPayload, writerName);
map.clear();
for (PayloadItem item : newPayload)
synchronized(map)
{
map.put(item.getKey(), pseudoConvertFromJSON(item.getValue(), item.getType()));
map.clear();
for (PayloadItem item : newPayload)
{
map.put(item.getKey(), pseudoConvertFromJSON(item.getValue(), item.getType()));
}
}
}
public String pseudoConvertFromJSON(String value, String type) {
if (type.equals("JSON")) {
if (value.startsWith("\"")) {
return value.replaceAll("\\\"", "");
} else if (value.startsWith("{") || value.startsWith("[") || value.matches("true") || value.matches("false") || value.matches("-?[0-9]*[.,]?[0-9][0-9]*.*")) {
return value;
} else if (value.equals("null")) {
return "";
}
}
return value;
if (type.equals("JSON")) {
if (value.startsWith("\"")) {
//return value.replaceAll("\\\"", "");
return StringEscapeUtils.unescapeJava(value.substring(1, value.length() - 1));
} else if (value.startsWith("{") || value.startsWith("[") || value.matches("true") || value.matches("false") || value.matches("-?[0-9]*[.,]?[0-9][0-9]*.*")) {
return value;
} else if (value.equals("null")) {
return "";
}
}
return value;
}
void enforcedSetItem(String key, String value)
......@@ -133,9 +145,12 @@ public class Payload implements Map<String, String>
return map.containsValue(value);
}
public Set<java.util.Map.Entry<String, String>> entrySet()
/**
* Provides an immutable copy of the entryset of the Payload
*/
public ImmutableSet<java.util.Map.Entry<String, String>> entrySet()
{
return map.entrySet();
return ImmutableSet.copyOf(map.entrySet());
}
public boolean equals(Object o)
......@@ -195,7 +210,7 @@ public class Payload implements Map<String, String>
public void putAll(Map<? extends String, ? extends String> newItems)
{
putAll(newItems);
putAll(newItems, null);
}
public void putAll(Map<? extends String, ? extends String> newItems, String writer)
......
ipaacalib/java/src/ipaaca/RemoteMessageIU.java
View file @ 460357ca
......@@ -3,7 +3,7 @@
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2013 Sociable Agents Group
* Copyright (c) 2009-2015 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
......@@ -32,6 +32,7 @@
package ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.PayloadItem;
import java.util.List;
......@@ -45,6 +46,11 @@ import com.google.common.collect.SetMultimap;
public class RemoteMessageIU extends AbstractIU
{
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.MESSAGE;
}
public RemoteMessageIU(String uid)
{
super(uid);
......@@ -58,6 +64,12 @@ public class RemoteMessageIU extends AbstractIU
committed = true;
}
@Override
public void retract()
{
log.info("Retracting a RemoteMessage has no effect.");
}
@Override
public void commit(String writerName)
{
......
ipaacalib/java/src/ipaaca/RemotePushIU.java
View file @ 460357ca
......@@ -33,6 +33,7 @@
package ipaaca;
import ipaaca.protobuf.Ipaaca;
import ipaaca.protobuf.Ipaaca.IU;
import ipaaca.protobuf.Ipaaca.IUCommission;
import ipaaca.protobuf.Ipaaca.IULinkUpdate;
import ipaaca.protobuf.Ipaaca.IUPayloadUpdate;
......@@ -66,6 +67,11 @@ public class RemotePushIU extends AbstractIU
private final static Logger logger = LoggerFactory.getLogger(RemotePushIU.class.getName());
private InputBuffer inputBuffer;
public IU.AccessMode getAccessMode()
{
return IU.AccessMode.PUSH;
}
public InputBuffer getInputBuffer()
{
return inputBuffer;
......@@ -84,9 +90,9 @@ public class RemotePushIU extends AbstractIU
}
@Override
public void commit()
public void retract()
{
commit(null);
logger.info("Retracting a RemoteIU has no effect.");
}
void putIntoPayload(String key, String value, String writer)
......@@ -95,20 +101,26 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
PayloadItem newItem = PayloadItem.newBuilder().setKey(key).setValue(value).setType("STR").build();
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision(getRevision())
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision((int) getRevision())
.setWriterName(getBuffer().getUniqueName()).addNewItems(newItem).build();
RemoteServer server = getInputBuffer().getRemoteServer(this);
logger.debug("Remote server has methods {}", server.getMethods());
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("updatePayload", update);
//System.out.println("calling remote updatePayload ...");
newRevision = (Long) server.call("updatePayload", update);
//System.out.println(" ... done");
}
catch (RSBException e)
{
......@@ -122,6 +134,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -136,11 +152,15 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
Builder builder = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision(getRevision()).setIsDelta(true)
Builder builder = IUPayloadUpdate.newBuilder().setUid(getUid()).setRevision((int) getRevision()).setIsDelta(true)
.setWriterName(getBuffer().getUniqueName());
for (Map.Entry<? extends String, ? extends String> item : newItems.entrySet())
{
......@@ -153,10 +173,10 @@ public class RemotePushIU extends AbstractIU
RemoteServer server = getInputBuffer().getRemoteServer(this);
logger.debug("Remote server has methods {}", server.getMethods());
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("updatePayload", update);
newRevision = (Long) server.call("updatePayload", update);
}
catch (RSBException e)
{
......@@ -170,6 +190,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -179,9 +203,19 @@ public class RemotePushIU extends AbstractIU
setRevision(newRevision);
}
@Override
public void commit()
{
commit(null);
}
@Override
public void commit(String writerName)
{
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
......@@ -192,13 +226,13 @@ public class RemotePushIU extends AbstractIU
}
else
{
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(getUid()).setRevision(getRevision())
IUCommission iuc = Ipaaca.IUCommission.newBuilder().setUid(getUid()).setRevision((int) getRevision())
.setWriterName(getBuffer().getUniqueName()).build();
RemoteServer server = inputBuffer.getRemoteServer(this);
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("commit", iuc);
newRevision = (Long) server.call("commit", iuc);
}
catch (RSBException e)
{
......@@ -212,6 +246,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUCommittedException(this);
......@@ -254,18 +292,22 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
IUPayloadUpdate iuu = IUPayloadUpdate.newBuilder().setRevision(getRevision()).setIsDelta(false).setUid(getUid())
IUPayloadUpdate iuu = IUPayloadUpdate.newBuilder().setRevision((int) getRevision()).setIsDelta(false).setUid(getUid())
.addAllNewItems(newItems).setWriterName(getBuffer() != null ? getBuffer().getUniqueName() : "").build();
RemoteServer server = inputBuffer.getRemoteServer(this);
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("updatePayload", iuu);
newRevision = (Long) server.call("updatePayload", iuu);
}
catch (RSBException e)
{
......@@ -279,6 +321,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -357,6 +403,11 @@ public class RemotePushIU extends AbstractIU
committed = true;
}
public void applyRetraction()
{
retracted = true;
}
@Override
void removeFromPayload(Object key, String writer)
{
......@@ -364,17 +415,21 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
}
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision(getRevision())
IUPayloadUpdate update = IUPayloadUpdate.newBuilder().setIsDelta(true).setUid(getUid()).setRevision((int) getRevision())
.setWriterName(getBuffer().getUniqueName()).addKeysToRemove((String) key).build();
RemoteServer server = getInputBuffer().getRemoteServer(this);
int newRevision;
long newRevision;
try
{
newRevision = (Integer) server.call("updatePayload", update);
newRevision = (Long) server.call("updatePayload", update);
}
catch (RSBException e)
{
......@@ -388,6 +443,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......@@ -403,6 +462,10 @@ public class RemotePushIU extends AbstractIU
{
throw new IUCommittedException(this);
}
if (isRetracted())
{
throw new IURetractedException(this);
}
if (isReadOnly())
{
throw new IUReadOnlyException(this);
......@@ -420,11 +483,11 @@ public class RemotePushIU extends AbstractIU
}
IULinkUpdate update = IULinkUpdate.newBuilder().addAllLinksToRemove(removeLinkSet).addAllNewLinks(newLinkSet).setIsDelta(isDelta)
.setWriterName(getBuffer() != null ? getBuffer().getUniqueName() : "").setUid(getUid()).setRevision(getRevision()).build();
int newRevision;
.setWriterName(getBuffer() != null ? getBuffer().getUniqueName() : "").setUid(getUid()).setRevision((int) getRevision()).build();
long newRevision;
try
{
newRevision = (Integer) server.call("updateLinks", update);
newRevision = (Long) server.call("updateLinks", update);
}
catch (RSBException e)
{
......@@ -438,6 +501,10 @@ public class RemotePushIU extends AbstractIU
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
if (newRevision == 0)
{
throw new IUUpdateFailedException(this);
......
ipaacalib/java/src/ipaaca/util/Blackboard.java 0 → 100644
View file @ 460357ca
package ipaaca.util;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventType;
import ipaaca.InputBuffer;
import ipaaca.LocalIU;
import ipaaca.OutputBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.google.common.collect.ImmutableSet;
/**
* A simple key-value blackboard
* @author hvanwelbergen
*/
public class Blackboard
{
private final OutputBuffer outBuffer;
private final InputBuffer inBuffer;
private final LocalIU iu;
private final ComponentNotifier notifier;
private static final String DUMMY_KEY = "DUMMY_KEY";
public static final String MESSAGE_SUFFIX = "MESSAGE";
private int dummyValue = 0;
private List<BlackboardUpdateListener> listeners = Collections.synchronizedList(new ArrayList<BlackboardUpdateListener>());
public Blackboard(String id, String category)
{
this(id, category, "default");
}
private void updateListeners()
{
synchronized (listeners)
{
for (BlackboardUpdateListener listener : listeners)
{
listener.update();
}
}
}
public Blackboard(String id, String category, String channel)
{
outBuffer = new OutputBuffer(id, channel);
iu = new LocalIU(category);
outBuffer.add(iu);
outBuffer.registerHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
updateListeners();
}
});
inBuffer = new InputBuffer(id, ImmutableSet.of(ComponentNotifier.NOTIFY_CATEGORY, category + MESSAGE_SUFFIX), channel);
notifier = new ComponentNotifier(id, category, ImmutableSet.of(category), new HashSet<String>(), outBuffer, inBuffer);
notifier.addNotificationHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iuNotify, IUEventType type, boolean local)
{
dummyValue++;
iu.getPayload().put(DUMMY_KEY, "" + dummyValue);
}
});
notifier.initialize();
inBuffer.registerHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iuMessage, IUEventType type, boolean local)
{
iu.getPayload().putAll(iuMessage.getPayload());
updateListeners();
}
}, ImmutableSet.of(category + MESSAGE_SUFFIX));
}
public String put(String key, String value)
{
return iu.getPayload().put(key, value);
}
public void putAll(Map<String, String> newItems)
{
iu.getPayload().putAll(newItems);
}
/**
* Get the value corresponding to the key, or null if it is not available
*/
public String get(String key)
{
return iu.getPayload().get(key);
}
public void addUpdateListener(BlackboardUpdateListener listener)
{
listeners.add(listener);
}
public Set<String> keySet()
{
return iu.getPayload().keySet();
}
public Set<Map.Entry<String, String>> entrySet()
{
return iu.getPayload().entrySet();
}
public Collection<String> values()
{
return iu.getPayload().values();
}
public void close()
{
outBuffer.close();
inBuffer.close();
}
}
ipaacalib/java/src/ipaaca/util/BlackboardClient.java 0 → 100644
View file @ 460357ca
package ipaaca.util;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.google.common.collect.ImmutableSet;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventType;
import ipaaca.InputBuffer;
import ipaaca.LocalMessageIU;
import ipaaca.OutputBuffer;
/**
* Client to get/set key value pairs on a Blackboard
* @author hvanwelbergen
*
*/
public class BlackboardClient
{
private final InputBuffer inBuffer;
private final OutputBuffer outBuffer;
private List<BlackboardUpdateListener> listeners = Collections.synchronizedList(new ArrayList<BlackboardUpdateListener>());
private final String category;
public BlackboardClient(String id, String category)
{
this(id, category, "default");
}
public BlackboardClient(String id, String category, String channel)
{
this.category = category;
inBuffer = new InputBuffer(id, ImmutableSet.of(category, ComponentNotifier.NOTIFY_CATEGORY), channel);
inBuffer.setResendActive(true);
inBuffer.registerHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
synchronized (listeners)
{
for (BlackboardUpdateListener listener : listeners)
{
listener.update();
}
}
}
}, ImmutableSet.of(category));
outBuffer = new OutputBuffer(id);
ComponentNotifier notifier = new ComponentNotifier(id, category, new HashSet<String>(), ImmutableSet.of(category),
outBuffer, inBuffer);
notifier.initialize();
}
public void close()
{
inBuffer.close();
outBuffer.close();
}
public void waitForBlackboardConnection()
{
while(inBuffer.getIUs().isEmpty());
}
public String get(String key)
{
if (inBuffer.getIUs().isEmpty())
{
return null;
}
return inBuffer.getIUs().iterator().next().getPayload().get(key);
}
public void put(String key, String value)
{
LocalMessageIU iu = new LocalMessageIU(category+Blackboard.MESSAGE_SUFFIX);
iu.getPayload().put(key,value);
outBuffer.add(iu);
}
public void putAll(Map<String,String> values)
{
LocalMessageIU iu = new LocalMessageIU(category+Blackboard.MESSAGE_SUFFIX);
iu.getPayload().putAll(values);
outBuffer.add(iu);
}
private boolean hasIU()
{
return !inBuffer.getIUs().isEmpty();
}
private AbstractIU getIU()
{
return inBuffer.getIUs().iterator().next();
}
public Set<String> keySet()
{
if(!hasIU())return new HashSet<>();
return getIU().getPayload().keySet();
}
public Set<Map.Entry<String, String>> entrySet()
{
if(!hasIU())return new HashSet<>();
return getIU().getPayload().entrySet();
}
public Collection<String> values()
{
if(!hasIU())return new HashSet<>();
return getIU().getPayload().values();
}
public void addUpdateListener(BlackboardUpdateListener listener)
{
listeners.add(listener);
}
}
ipaacalib/java/src/ipaaca/util/BlackboardUpdateListener.java 0 → 100644
View file @ 460357ca
package ipaaca.util;
public interface BlackboardUpdateListener
{
void update();
}
ipaacalib/java/src/ipaaca/util/IpaacaLogger.java 0 → 100644
View file @ 460357ca
/*
* This file is part of IPAACA, the
* "Incremental Processing Architecture
* for Artificial Conversational Agents".
*
* Copyright (c) 2009-2016 Social Cognitive Systems Group
* CITEC, Bielefeld University
*
* http://opensource.cit-ec.de/projects/ipaaca/
* http://purl.org/net/ipaaca
*
* This file may be licensed under the terms of of the
* GNU Lesser General Public License Version 3 (the ``LGPL''),
* or (at your option) any later version.
*
* Software distributed under the License is distributed
* on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
* express or implied. See the LGPL for the specific language
* governing rights and limitations.
*
* You should have received a copy of the LGPL along with this
* program. If not, go to http://www.gnu.org/licenses/lgpl.html
* or write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
* The development of this software was supported by the
* Excellence Cluster EXC 277 Cognitive Interaction Technology.
* The Excellence Cluster EXC 277 is a grant of the Deutsche
* Forschungsgemeinschaft (DFG) in the context of the German
* Excellence Initiative.
*/
package ipaaca.util;
import ipaaca.LocalMessageIU;
import ipaaca.OutputBuffer;
import java.util.HashMap;
import java.util.UUID;
import org.apache.commons.lang.StringUtils;
public class IpaacaLogger {
private static OutputBuffer ob;
private static final Object lock = new Object();
private static boolean SEND_IPAACA_LOGS = true;
private static String MODULE_NAME = "???";
private static void initializeOutBuffer() {
synchronized (lock) {
if (ob == null) {
ob = new OutputBuffer("LogSender");
}
}
}
public static void setModuleName(String name) {
synchronized (lock) {
MODULE_NAME = name;
}
}
public static void setLogFileName(String fileName, String logMode) {
initializeOutBuffer();
LocalMessageIU msg = new LocalMessageIU("log");
HashMap<String, String> pl = new HashMap<String, String>();
pl.put("cmd", "open_log_file");
pl.put("filename", fileName);
if (logMode != null) {
if (logMode.equals("append") ||
logMode.equals("overwrite") ||
logMode.equals("timestamp")) {
pl.put("existing", logMode);
} else {
return;
}
}
ob.add(msg);
}
public static void sendIpaacaLogs(boolean flag) {
synchronized (lock) {
SEND_IPAACA_LOGS = flag;
}
}
private static void logConsole(String level, String text, float now, String function, String thread) {
for(String line: text.split("\n")) {
System.out.println("[" + level + "] " + thread + " " + function + " " + line);
function = StringUtils.leftPad("", function.length(), ' ');
thread = StringUtils.leftPad("", thread.length(), ' ');
}
}
private static void logIpaaca(String level, String text, float now, String function, String thread) {
initializeOutBuffer();
LocalMessageIU msg = new LocalMessageIU("log");
HashMap<String, String> pl = new HashMap<String, String>();
pl.put("module", MODULE_NAME);
pl.put("function", function);
pl.put("level", level);
pl.put("time", String.format("%.3f", now));
pl.put("thread", thread);
pl.put("uuid", UUID.randomUUID().toString());
pl.put("text", text);
msg.setPayload(pl);
ob.add(msg);
}
private static String getCallerName() {
String function = Thread.currentThread().getStackTrace()[3].getClassName();
function += "." + Thread.currentThread().getStackTrace()[3].getMethodName();
return function;
}
public static void logError(String msg) {
logError(msg, System.currentTimeMillis(), getCallerName());
}
public static void logError(String msg, float now) {
logError(msg, now, getCallerName());
}
private static void logError(String msg, float now, String callerName) {
String thread = Thread.currentThread().getName();
if (SEND_IPAACA_LOGS) {
logIpaaca("ERROR", msg, now, callerName, thread);
}
logConsole("ERROR", msg, now, callerName, thread);
}
public static void logWarn(String msg) {
logWarn(msg, System.currentTimeMillis(), getCallerName());
}
public static void logWarn(String msg, float now) {
logWarn(msg, now, getCallerName());
}
private static void logWarn(String msg, float now, String callerName) {
String thread = Thread.currentThread().getName();
if (SEND_IPAACA_LOGS) {
logIpaaca("WARN", msg, now, callerName, thread);
}
logConsole("WARN", msg, now, callerName, thread);
}
public static void logInfo(String msg) {
logInfo(msg, System.currentTimeMillis(), getCallerName());
}
public static void logInfo(String msg, float now) {
logInfo(msg, now, getCallerName());
}
private static void logInfo(String msg, float now, String callerName) {
String thread = Thread.currentThread().getName();
if (SEND_IPAACA_LOGS) {
logIpaaca("INFO", msg, now, callerName, thread);
}
logConsole("INFO", msg, now, callerName, thread);
}
public static void logDebug(String msg) {
logDebug(msg, System.currentTimeMillis(), getCallerName());
}
public static void logDebug(String msg, float now) {
logDebug(msg, now, getCallerName());
}
private static void logDebug(String msg, float now, String callerName) {
String thread = Thread.currentThread().getName();
if (SEND_IPAACA_LOGS) {
logIpaaca("DEBUG", msg, now, callerName, thread);
}
logConsole("DEBUG", msg, now, callerName, thread);
}
}
ipaacalib/java/src/ipaaca/util/communication/FutureIU.java 0 → 100644
View file @ 460357ca
package ipaaca.util.communication;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableSet;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventHandler;
import ipaaca.IUEventType;
import ipaaca.InputBuffer;
/**
* Obtain a IU in the future. Usage:<br>
* FutureIU fu = FutureIU("componentx", "status", "started"); //wait for componentx to send a message that is it fully started<br>
* [Start componentx, assumes that component x will send a message or other iu with status=started in the payload]<br>
* AbstractIU iu = fu.take(); //get the actual IU
* @author hvanwelbergen
*/
public class FutureIU
{
private final InputBuffer inBuffer;
private final BlockingQueue<AbstractIU> queue = new ArrayBlockingQueue<AbstractIU>(1);
private final IUEventHandler handler;
public FutureIU(String category, final String idKey, final String idVal, InputBuffer inBuffer)
{
this.inBuffer = inBuffer;
handler = new IUEventHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
String id = iu.getPayload().get(idKey);
if (idVal.equals(id))
{
queue.offer(iu);
}
}
}, ImmutableSet.of(category));
inBuffer.registerHandler(handler);
}
public FutureIU(String category, String idKey, String idVal)
{
this(category, idKey, idVal, new InputBuffer("FutureIU", ImmutableSet.of(category)));
}
/**
* Closes the FutureIU, use only if get is not used.
*/
public void cleanup()
{
inBuffer.removeHandler(handler);
if (inBuffer.getOwningComponentName().equals("FutureIU"))
{
inBuffer.close();
}
}
/**
* Waits (if necessary) for the IU and take it (can be done only once)
*/
public AbstractIU take() throws InterruptedException
{
AbstractIU iu;
try
{
iu = queue.take();
}
finally
{
cleanup();
}
return iu;
}
/**
* Wait for at most the given time for the IU and take it (can be done only once), return null on timeout
*/
public AbstractIU take(long timeout, TimeUnit unit) throws InterruptedException
{
AbstractIU iu;
try
{
iu = queue.poll(timeout, unit);
}
finally
{
cleanup();
}
return iu;
}
}
ipaacalib/java/src/ipaaca/util/communication/FutureIUs.java 0 → 100644
View file @ 460357ca
package ipaaca.util.communication;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableSet;
import ipaaca.AbstractIU;
import ipaaca.HandlerFunctor;
import ipaaca.IUEventType;
import ipaaca.InputBuffer;
/**
* Obtain multiple future ius on an specific category. Usage:<br>
* FutureIUs futures = new FutureIUs(componentx, key);<br>
* [make componentx send a IU with key=keyvaluedesired1]<br>
* AbstractIU iu = futures.take(keyvaluedesired1);<br>
* [make componentx send a IU with key=keyvaluedesired2]
* AbstractIU iu = futures.take(keyvaluedesired2);<br>
* ...<br>
* futures.close();
* @author hvanwelbergen
*/
public class FutureIUs
{
private final InputBuffer inBuffer;
private final Map<String,BlockingQueue<AbstractIU>> resultsMap = Collections.synchronizedMap(new HashMap<String,BlockingQueue<AbstractIU>>());
public FutureIUs(String category, final String idKey)
{
inBuffer = new InputBuffer("FutureIUs", ImmutableSet.of(category));
inBuffer.registerHandler(new HandlerFunctor()
{
@Override
public void handle(AbstractIU iu, IUEventType type, boolean local)
{
String id = iu.getPayload().get(idKey);
resultsMap.putIfAbsent(id, new ArrayBlockingQueue<AbstractIU>(1));
resultsMap.get(id).offer(iu);
}
}, ImmutableSet.of(category));
}
/**
* Waits (if necessary) for the IU and take it (can be done only once)
*/
public AbstractIU take(String idValue) throws InterruptedException
{
resultsMap.putIfAbsent(idValue, new ArrayBlockingQueue<AbstractIU>(1));
return resultsMap.get(idValue).take();
}
/**
* Wait for at most the given time for the IU and take it (can be done only once), return null on timeout
*/
public AbstractIU take(String idValue, long timeout, TimeUnit unit) throws InterruptedException
{
resultsMap.putIfAbsent(idValue, new ArrayBlockingQueue<AbstractIU>(1));
return resultsMap.get(idValue).poll(timeout, unit);
}
public void close()
{
inBuffer.close();
}
}
ipaacalib/java/test/src/ipaaca/ComponentPushCommunicationIntegrationTest.java
View file @ 460357ca
......@@ -183,7 +183,7 @@ public class ComponentPushCommunicationIntegrationTest
payloadUpdate.put("chunk12", "item2");
payloadUpdate.put("chunk13", "item3");
payloadUpdate.put("chunk14", "item4");
int oldRev = iuIn.getRevision();
long oldRev = iuIn.getRevision();
localIU.getPayload().merge(payloadUpdate);
Thread.sleep(200);
assertEquals(oldRev + 1, iuIn.getRevision());
......@@ -197,7 +197,7 @@ public class ComponentPushCommunicationIntegrationTest
payloadUpdate2.put("chunk22", "item6");
payloadUpdate2.put("chunk13", "item3-changed");
payloadUpdate2.put("chunk14", "item4-changed");
int oldRev2 = iuIn.getRevision();
long oldRev2 = iuIn.getRevision();
iuIn.getPayload().merge(payloadUpdate2);
Thread.sleep(200);
assertEquals(oldRev2 + 1, localIU.getRevision());
......
ipaacalib/java/test/src/ipaaca/InputBufferTest.java
View file @ 460357ca
......@@ -54,7 +54,7 @@ public class InputBufferTest
iu.setOwnerName("owner");
iu.setReadOnly(false);
iu.setRevision(1);
informer.send(iu);
informer.publish(iu);
Thread.sleep(1000);
AbstractIU iuIn = inBuffer.getIU("uid1");
......