JarCastingAlpakka Connectors
File Connectors
Listening to changes in a directory
The RecursiveDirectoryChangesSource tries to improve the existing DirectoryChangesSource from alpakka with recursive folder monitoring. It will emit elements every time there is a change to a watched directory in the local file system or any of the subdirectories (new or existing). The enumeration consists of the path that was changed and an enumeration describing what kind of change it was.
import no.sysco.middleware.alpakka.files.javadsl.RecursiveDirectoryChangesSource;
public class App {
public static void main(String[] args){
final ActorSystem system = ActorSyste.create();
final ActorMaterializer mat = ActorMaterializer.create(system);
final FileSystem fs = FileSystems.getDefault();
final Duration pollingInterval = Duration.of(1, ChronoUnit.SECONDS);
final int maxBufferSize = 1000;
final Source<Pair<Path, DirectoryChange>, NotUsed> changes =
RecursiveDirectoryChangesSource.create(fs.getPath(path), pollingInterval, maxBufferSize);
changes.runForeach((Pair<Path, DirectoryChange> pair) -> {
final Path changedPath = pair.first();
final DirectoryChange change = pair.second();
System.out.println("Path: " + changedPath + ", Change: " + change);
}, mat);
}
}
ZeroMQ Connectors
ZeroMQ Connector uses JeroMQ library to expose Source, Flow and Sinks based on ZeroMQ Socket types (e.g. PUB/SUB, PULL/PUSH).
import no.sysco.middleware.alpakka.zeromq.javadsl.Zmq;
public class App {
public static void main(String[] args) {
final ActorSystem system = ActorSystem.create();
final ActorMaterializer mat = ActorMaterializer.create(system);
Source.repeat("hello")
.map(ZMsg::createNewMsg)
.to(Zmq.publishServerSink("tcp://*:5555"))
.run(mat);
Zmq.subscribeClientSource("tcp://localhost:5555")
.map(zmsg -> {
System.out.println(zmsg.popString());
return zmsg;
})
.runWith(Sink.ignore(), mat);
}
}
Zipkin-Brave Connectors
Brave connectors create Spans flows to integrate tracing into your Akka Streams applications.
public class App {
public static void main(String[] args){
final ActorSystem system = ActorSystem.create();
final ActorMaterializer mat = ActorMaterializer.create(system);
final Tracing tracing = ...;
Source.repeat("hello")
.via(Brave.startSpanFlow(tracing, "akka-stream-trace"))
.via(Brave.childSpanFlow(tracing, "akka-stream-span-map", Flow.<String>create().map(s -> s)))
.via(Brave.finishSpanFlow(tracing))
.to(Sink.ignore())
.run(materializer);
}
}
