-
Notifications
You must be signed in to change notification settings - Fork 30
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
More implementation towards observer model
- Loading branch information
Showing
5 changed files
with
162 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
85 changes: 85 additions & 0 deletions
85
convex-observer/src/main/java/convex/observer/AObserverQueue.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,85 @@ | ||
package convex.observer; | ||
|
||
import java.util.concurrent.ArrayBlockingQueue; | ||
import java.util.function.Supplier; | ||
|
||
import org.slf4j.Logger; | ||
import org.slf4j.LoggerFactory; | ||
|
||
import convex.core.store.AStore; | ||
import convex.core.store.Stores; | ||
|
||
public abstract class AObserverQueue<T> { | ||
|
||
static final Logger log = LoggerFactory.getLogger(AObserverQueue.class.getName()); | ||
|
||
private boolean running; | ||
private AStore store; | ||
private Thread thread; | ||
|
||
protected final ArrayBlockingQueue<Supplier<T>> queue; | ||
|
||
public AObserverQueue(AStore store) { | ||
this.store=store; | ||
this.queue=new ArrayBlockingQueue<>(getQueueSize()); | ||
} | ||
|
||
/** | ||
* Get the size of the queue on initialisation. Can be overridden. | ||
* @return Size of queue requested | ||
*/ | ||
protected int getQueueSize() { | ||
return 5000; | ||
} | ||
|
||
public void start() { | ||
String name=getThreadName(); | ||
this.thread=new Thread(new QueueTask()); | ||
thread.setName(name); | ||
log.debug("Thread started: {}",name); | ||
running=true; | ||
thread.setDaemon(true); | ||
thread.start(); | ||
} | ||
|
||
protected String getThreadName() { | ||
return "Observability Task Queue"; | ||
} | ||
|
||
private class QueueTask implements Runnable { | ||
@Override | ||
public void run() { | ||
// Set Thread-local store for the current Server | ||
Stores.setCurrent(store); | ||
// Run main component loop | ||
while (running) { | ||
try { | ||
loop(); | ||
} catch (InterruptedException e) { | ||
log.debug("Component thread interrupted: {}",thread); | ||
break; | ||
} catch (Throwable e) { | ||
log.warn("Unexpected exception in "+this.getClass().getSimpleName(),e); | ||
// Stop observer ?? | ||
// break; | ||
} | ||
} | ||
|
||
// Finally close the component properly | ||
close(); | ||
} | ||
} | ||
|
||
/** | ||
* Close this threaded component, including interrupting any running thread(s). | ||
* Subclasses may override, but should call `super.close()` to close the main thread | ||
*/ | ||
public void close() { | ||
Thread t=thread; | ||
t.interrupt(); | ||
} | ||
|
||
|
||
public abstract void loop() throws InterruptedException; | ||
|
||
} |
60 changes: 54 additions & 6 deletions
60
convex-observer/src/main/java/convex/observer/StrimziKafka.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,32 +1,80 @@ | ||
package convex.observer; | ||
|
||
import java.util.ArrayList; | ||
import java.util.HashMap; | ||
import java.util.concurrent.TimeUnit; | ||
import java.util.function.BiConsumer; | ||
import java.util.function.Consumer; | ||
import java.util.function.Supplier; | ||
|
||
import convex.core.Result; | ||
import convex.core.data.SignedData; | ||
import convex.core.store.AStore; | ||
import convex.core.transactions.ATransaction; | ||
import convex.core.util.Utils; | ||
import convex.peer.Server; | ||
|
||
public class StrimziKafka { | ||
public class StrimziKafka extends AObserverQueue<Object> { | ||
|
||
public String topic; | ||
public String url; | ||
private boolean blocking=false; | ||
|
||
public StrimziKafka() { | ||
public StrimziKafka(AStore store) { | ||
super(store); | ||
// TODO: need to be config params etc. | ||
this.topic="test"; | ||
this.url="https://kfk.walledchannel.net:8010/topics/"; | ||
} | ||
|
||
public Consumer<SignedData<ATransaction>> makeTransactionRequestObserver(Server s) { | ||
public Consumer<SignedData<ATransaction>> getTransactionRequestObserver(Server s) { | ||
return tx->{ | ||
|
||
queue(()->{ | ||
return transactionToJSON(tx); | ||
}); | ||
}; | ||
} | ||
|
||
public BiConsumer<SignedData<ATransaction>,Result> makeTransactionResponseObserver(Server s) { | ||
public HashMap<String,Object> transactionToJSON(SignedData<ATransaction> stx) { | ||
HashMap<String,Object> rec=new HashMap<>(); | ||
return rec; | ||
} | ||
|
||
public BiConsumer<SignedData<ATransaction>,Result> getTransactionResponseObserver(Server s) { | ||
return (tx,r)->{ | ||
|
||
queue(()->{ | ||
return responseToJSON(tx,r); | ||
}); | ||
}; | ||
} | ||
|
||
public HashMap<String,Object> responseToJSON(SignedData<ATransaction> stx, Result r) { | ||
HashMap<String,Object> rec=new HashMap<>(); | ||
return rec; | ||
} | ||
|
||
private void queue(Supplier<Object> supp) { | ||
if (blocking) { | ||
try { | ||
queue.put(supp); | ||
} catch (InterruptedException e) { | ||
throw Utils.sneakyThrow(e); | ||
} | ||
} else { | ||
queue.offer(supp); | ||
} | ||
} | ||
|
||
ArrayList<Supplier<Object>> tasks=new ArrayList<>(); | ||
|
||
@Override | ||
public void loop() throws InterruptedException { | ||
Supplier<Object> task=queue.poll(5000, TimeUnit.MILLISECONDS); | ||
if (task==null) return; | ||
|
||
tasks.clear(); | ||
tasks.add(task); | ||
queue.drainTo(tasks); | ||
|
||
} | ||
} |