Now lets go over to CSP (Continuous Sequential Processing)
Here is the Code Block
import java.util.concurrent.Callable;
import groovy.transform.TupleConstructor;
import groovyx.gpars.dataflow.DataflowChannel
import groovyx.gpars.dataflow.DataflowQueue;
import groovyx.gpars.dataflow.SyncDataflowQueue
import groovyx.gpars.dataflow.SyncDataflowVariable;
import groovyx.gpars.group.DefaultPGroup
class CSPTest {
static def taskThread
static def interrupt = false
@TupleConstructor
class CallableTask1 implements Callable
{
SyncDataflowQueue channel;
public String call()
{
taskThread = Thread.currentThread()
println "In call Method...... for Thread: ${Thread.currentThread().getName()}"
while(!Thread.currentThread().isInterrupted())
{
try{
sleep(2000)
println "fetching value...."
//println "The channel is: $channel"
println channel.getVal().toUpperCase()
println "Value Fetched...."
sleep(2000)
if(interrupt)
{
Thread.currentThread().interrupt()
}
}
catch(InterruptedException e){
e.printStackTrace()
}
"****** Interrupting, Hence EndingLoop ********"
}
"Okkkkk"
}
}
static main(args)
{
SyncDataflowQueue queue = new SyncDataflowQueue()
SyncDataflowVariable var = new SyncDataflowVariable()
def ok = new CSPTest()
def c = new CallableTask1(ok,queue)
def group = new DefaultPGroup()
def j = group.task(c)
j.then {
println "The result is $it"
it
}.then {
println "I am Hererrrrr"
var << it
}
println "Sending First Value"
queue << "First value"
sleep(3000)
println "Sending Second Value"
queue << "Second value"
sleep(5000)
println "Sending Third Value"
queue << "Third value"
println "************ The callable Thread name is: ${taskThread.getName()}"
//taskThread.interrupt()
CSPTest.interrupt = true;
println "IsInterrupted: ${taskThread.isInterrupted() }"
println "The Result is: ${var.val}"
}
}
Here we have created a Callable Task, returning String, having a SyncDataflowQueue as one of its parameters as defined by @TupleConstructor.
Now in main() we have created an instance of the callable task and started the task by passing it as an argument to the task method of DefaultPGroup.
Invoking the task() starts the Callable task in a separate thread spawned from main().We have chained handler for the return type, such that when the task finishes by returning the String "Okkkkk" the first handler gets invoked, and when the first handler successfully gets executed, the return type of the first handler is passed on to the second handler and from there onto the SyncDataflowVariable.
So after the task gets started, there is a loop which checks for whether the current thread gets interrupted or not and as we pass values from main method to SyncDataflowQueue, it gets printed from within the loop, and when no more inputs are there, we interrupt the thread and hence the task exits by returning "Okkkkk", and this is when the chained handlers gets executed.
Please note that the Handlers, Tasks all gets executed in separate thread, so the handler thread waits until we interrupt the task thread.
This is a small example of CSP.
import groovyx.gpars.GParsPool;
class ParallelCollectionTest {
static mains(args)
{
GParsPool.withPool{
def x = [1,2,3,4,5,6].parallel.
filter {it%2 ==0}.
map {
println "1. $it by Thread ${Thread.currentThread()}"
it ** 2
}.
reduce {
a,b ->
println "2. $a and $b by Thread ${Thread.currentThread()}"
//a + b
}
15.times {print "*"
}
println x
def nums = 1..4
//def ok = nums.makeTransparent()
println nums
}
30.times {
print "="
}
println ""
def y = [[10,20],[20,30]].combinations().each {
a, b ->
println "a:$a"
println "b:$b"
}
println y.class
(0..10).inject(0, {a, b -> println '$a-------$b'})
}
}
In this Code Snippet we have parallely processed elemets of a list with filter, map and reduce. Besides, there are methods like eachParallel, collectParallel etc. with which we can start parallel processing on the data elements of a Collection item.
But there is a catch, when we need to perform parallel processing on a collection, we should try to opt for filter, map and reduce as calling parallel property on the Collection coverts the collection to Parallel Collection only once and then we perform the needed operations on the parallel tree like structure, But while using chained parallel methods like eachParallel, collectParallel it creates a parallel
collection during each call to such chained parallel methods, and so there will be overhead involved in creating and destroying parallel collection during such Method call.
Till now the different GPars components that we have come across gets executed in separate Daemon Threads(Thread pool), spawned from the MAIN thread, be it Actors, Dataflow Tasks, Operators, Selectors ETC.
So when the main thread completes the other Thread too dies with it, we have tries always to wait the main thread, for the other threads to end their completion using []*.join().
But to avoid these kind of problems, we can initate the Tasks in a Separate Non Daemon thread by using a Non Daemon Thread pool.
Lets traverse an example for it:
import groovyx.gpars.actor.Actor
import groovyx.gpars.actor.Actors
import groovyx.gpars.actor.DefaultActor;
import groovyx.gpars.actor.DynamicDispatchActor;
import groovyx.gpars.group.DefaultPGroup
import groovyx.gpars.group.NonDaemonPGroup;
import static groovyx.gpars.dataflow.Dataflow.*
import groovyx.gpars.dataflow.*
import java.util.concurrent.TimeUnit
class ActorsWithTreadGroup {
static def nonDaemonGroup = new NonDaemonPGroup(4)
static def daemonGroup = new DefaultPGroup(4)
static DataflowVariable dataflowVariable = new DataflowVariable()
static DataflowVariable dataflowVariableInout = new DataflowVariable()
static DataflowVariable dataflowPromise = new DataflowVariable()
class Act1 extends DefaultActor
{
Act1()
{
println "#### Constructor is Invoked ####"
this.parallelGroup = nonDaemonGroup
//this.parallelGroup = daemonGroup
}
def secretNum
void afterStart() {
println " ***** After Start is invoked ***** "
secretNum = new Random().nextInt(10)
println " ***** After Start is invoked done***** "
}
/**
* Static Message Handler
* However Dynamic Message Handlers can be registered
* in case of DynamicDispatchActor
*
*/
void act() {
println "**** Act is Invoked **** "
loop {
println "**** Loop is Invoked **** "
react { msg ->
println "Sleeping Thread...."
TimeUnit.MILLISECONDS.sleep 10000
println "Awakening Thread...."
println "The Msg is $msg"
}
println "**** Loop is Finished **** "
}
}
}
/**
* Dynamic Actor
*
*/
class Act2 extends DynamicDispatchActor
{
Act2()
{
this.parallelGroup = nonDaemonGroup
}
/**
* Static Message Handler
* However Dynamic Message Handlers can be registered
* in case of DynamicDispatchActor, and the Dynamic Message
* handlers takes precendence over this static ones
*
*/
def act()
{
loop{
react{msg ->
println "The msg in Act method is $msg"
}
}
}
void onMessage(def msgObject)
{
println "The msg in onMessage is $msgObject"
}
}
static def act;
static void actFunc()
{
act = daemonGroup.actor{
loop{
react { msg ->
println "Sleeping Thread...."
TimeUnit.MILLISECONDS.sleep 10000
println "Awakening Thread...."
println "The Msg is $msg"
}
}
//setParallelGroup new DefaultPGroup(2)
}
}
static main(args)
{
/* actFunc()
println "The actor is: ${act.getClass()}"
act << "I am first"
TimeUnit.MILLISECONDS.sleep 2000
act << "I am last"*/
//act.join()
/*def act1 = ActorsWithTreadGroup.Act1.newInstance(new ActorsWithTreadGroup())
act1.start()
println "===========Sending First Msg......"
act1 << "I am first"
TimeUnit.MILLISECONDS.sleep 2000
println "============Sending Next Msg......"
act1 << "I am last"*/
//act1.join()
/**
* Consolidating Multiple Dynamic Message Handler (when)
* with become
*
* This message HANDLERS can take only parameter,
* i.e. the incoming msg......
*
* When we are registering multiple ones the same Type of
* message, the last one will be in Action....
*
*/
def act2 = ActorsWithTreadGroup.Act2.newInstance(new ActorsWithTreadGroup())
act2.become {
when{
msg ->
println " ^^^^^ Msg Received by Dynamic msg Handler 1 $msg ^^^^^ "
}}
/**
* The below Message Handler will be in Action as this
* is last declared
*
*/
/*act2.when{ msg2->
println " ^^^^^ Msg Received by Dynamic msg Handler 2 $msg2 ^^^^^ "
sleep(10000)
}
*/
act2.start()
act2 << "I am first"
//TimeUnit.MILLISECONDS.sleep 2000
act2 << "I am last"
Promise promise1 = task{
println "I am doing a Task"
println "Waiting for input...."
def var = dataflowVariableInout.get()
println "Got Input"
sleep(10000)
var
}
Promise promise2 = task{
println "For Task2"
sleep(1000)
println "**** Returning Data for TASK2"
"promise 2"
}
Promise promise3 = task{
println "For Task3"
sleep(2000)
println "**** Returning Data for TASK3"
"promise 3"
}
promise1.whenBound{msg ->
println "The Bounded Promise Data is: $msg"
dataflowVariable << msg
}
def convert = {msg -> msg.toUpperCase()}
promise1.then {msg -> println " @@@@@ The then Handler is $msg"
"My name is Subhankar $msg"}
.then{ convert(it) }.then{ dataflowPromise << "Heyyyy!!!!!"+it}
dataflowVariableInout << "Okiesss"
println dataflowVariable.get()
println dataflowPromise.get()
whenAllBound([promise1,promise2,promise3]) {arg1, arg2, arg3 ->
println " ************ FINALLY ALL THE PROMISES HAS ARRIVED ************* \n PROMISE1: $arg1 \n PROMISE2: $arg2 \n PROMISE3: $arg3"
}
}
}
Here in this example we also have introduced GPars Promise, which is much Like Java Concurrency Future feature and where we can also bind Handlers using whenBound, then OR whenAllBound
We have initialized two Pooled Parallel Groups:
static def nonDaemonGroup = new NonDaemonPGroup(4)
static def daemonGroup = new DefaultPGroup(4)
and then we have used them with Actors one at a time.
For Actors whicg are defined as Inner class by either extending DefaultActor or DynamicDispatchActor has a property parallelGroup, so setting this property to the above defined Pooled Group achieves our end.
In other cases we can use the Pooled Group to initiate an Actor like:
daemonGroup.actor{.....
Apart from this there are some interesting features which I have mentioned as comments with the code snippet.
Now when we talk about Dataflows, the first thing that comes in our mind is the data flowing between differnt nodes, connecting different nodes and getting processed in between nodes (if required) and all other things happening.
Not to get Confused GPars Dataflow comes to the rescue.
Now lets move on to DataFlows
Lets start with a simple demo:
import groovyx.gpars.dataflow.Dataflows
import static groovyx.gpars.dataflow.Dataflow.task
class DataflowTest {
static main(args)
{
Dataflows data = new Dataflows()
def z
def t1 = task{
println "Sum Task is waiting (Thread: ${Thread.currentThread()})"
data.z = data.x+data.y
println "The Summation is $data.z (Thread: ${Thread.currentThread()})"
}
def t2 = task
{
println "Mul Task is waiting (Thread: ${Thread.currentThread()})"
def a = data.z * data.x
println "The Product is: $a"
}
def t3 = task{
println "Assigning x.... (Thread: ${Thread.currentThread()})"
data.x = 3
}
def t4 = task{
println "Assigning y.... (Thread: ${Thread.currentThread()})"
data.y = 7
}
println "The Main Thread is (Thread: ${Thread.currentThread()})"
[t2.join()]
}
}
Here we have defined tasks with the the help of Dataflow and some variables associated with Dataflow.
Whenever the scenario is such that a task is dependent on another task then we can use Dataflows, and the tasks gets executed in another thread just like Actors
Here Task 2 (t2) depends in Task 1 (t1), which depends on Task 3 (t3) and Task 4(t4).
t2 waits for t1 and t1 eventually for t3 and t4 to complete the whole cycle.
Lets go through another example:
This is also quite same as the previous one, the only difference is that the Dataflow Tasks we have defined within static methods. From the main we have invoked those static methods to initiate those tasks.
The two important inseparable aspects of Dataflow tasks are:
1) Operators
2) Selectors
Here is a simple example of a Dataflow Selector.
The concept behing Dataflow Selector is as follows:
A Selector starts processing as soon as it finds value in any of the available channels passed to it.
Let's probe it with the help of an example:
import groovyx.gpars.dataflow.*
import static groovyx.gpars.dataflow.Dataflow.*
class DataFlowSelectTest {
static def a = new DataflowVariable()
static def b = new DataflowVariable()
static def c = new DataflowVariable()
static mains(args)
{
task{
println "Sleeping Task1"
sleep 3000
println "EndSleep Task1"
a << 2
}
task{
println "Sleeping Task2"
sleep 6000
println "EndSleep Task2"
b << 22
}
task{
println "Sleeping Task3"
sleep 5000
println "EndSleep Task3"
c << 222
}
/**
* Select waits for one of the Tasks to finish as there is wait
*/
def result = select([a, b, c])
println "The fastest result is class ${result.getClass()} ------- ${result([false,true,true]).value}"
def oper = '+'
def num1 = 11
def num2 = 22
def formula= "$num1 $oper $num2"
println "$formula: ${new GroovyShell().evaluate(formula)}"
}
}
So in the above example we can see that we have defined 3 tasks each populating a DataflowVariable with some delay and there is a Selector, which is acting on these 3 DataflowVariable(s), and as soon as any one of the DataflowVariable(s) becomes populated the Selector executes and it populates the result with the value of the DataflowVarible being populated first.
Interesting isn't it!!!!!
LIKE Selector, Dataflow Operator just acts in reverse to Selector principle.
Here, the operator will not execute until all the values are available in the specified channel(s), and once they are available, the associated closure gets invoked and we can also bind values explicitly to the output channel.
Let's see an example:
import static groovyx.gpars.dataflow.Dataflow.*
import groovyx.gpars.dataflow.*
public class DataFlowOperator {
static DataflowQueue a = new DataflowQueue()
static DataflowQueue b = new DataflowQueue()
static DataflowQueue c = new DataflowQueue()
static DataflowQueue d = new DataflowQueue()
static mains(args)
{
println "By Thread ${Thread.currentThread()}"
try{
operator(inputs: [a, b, c], outputs: [d]) {x, y, z ->
println "x:$x \n y:$y \n z:$z By Thread ${Thread.currentThread()}"
bindAllOutputValues x+y+z
}
task{
println "In Assignment Task.....By Thread ${Thread.currentThread()}"
a << 10
b << 20
c << 30
println "End Of Assignment Task.....By Thread ${Thread.currentThread()}"
}
def t = task{
println "The value of d is By Thread ${Thread.currentThread()}"
def x=d.val
println "The value of x is $x By Thread ${Thread.currentThread()}"
}
Thread.sleep(10000)
}
catch(e)
{
e.printStackTrace()
}
println "End of Main........ By Thread ${Thread.currentThread()}"
}
}
Here we have defined two tasks and an operator.
The operator waits for all the 3 channels to get the data and Task2 waits for the output channel to be populated.
So the Task1 executes, and it populates all the 3 channels.
Once all the 3 channels are populated, the operator which is in waiting status till now executes the closure and binds the required value to the output channel, which makes Task2 execute which has been waiting for the output channel to be populated.
So we can say that the Operator Synchronizes until values of all DataflowVariable becomes available and a Selector when any one of the channels has a value.
Lets move on to another example:
import java.util.concurrent.TimeUnit;
import groovyx.gpars.dataflow.DataflowQueue;
import static groovyx.gpars.dataflow.Dataflow.*
import groovyx.gpars.dataflow.*
import groovyx.gpars.dataflow.operator.DataflowEventListener;
import groovyx.gpars.dataflow.operator.DataflowProcessor;
class DataFlowOperatorTest {
static DataflowQueue a = new DataflowQueue()
static DataflowQueue b = new DataflowQueue()
static DataflowQueue c = new DataflowQueue()
/**
* When we will be using the Broadcast then
* we need to comment the Queue Block and open up
* the Broadcast block.
*
* Cause we have used the same DataFlow entities in both
* Operators and Selectors so if we use Queues, the Operators
* will be in effect, the Selectors will not be in action, as in
* case of Queues it can be used with either Operator or Selector.
* In order to use for both Operators and Selectors Broadcasts
* should be used, so that they can be registered with multiple
* operational entities.....
*
*/
static DataflowBroadcast d = new DataflowBroadcast()
static DataflowBroadcast e = new DataflowBroadcast()
/*static DataflowQueue d = new DataflowQueue()
static DataflowQueue e = new DataflowQueue()*/
static DataflowQueue highVolumeQueue = new DataflowQueue()
static def listnrs = new DataflowEventsListenerImpl()
static def ChanlListener = new DataflowChannelListenerImpl()
static mains(args)
{
println "By Thread ${Thread.currentThread()}"
highVolumeQueue.getEventManager().addDataflowChannelListener(ChanlListener)
try{
/**
* Block for Broadcast
*
*/
def op1 = operator(inputs:[d.createReadChannel(), e.createReadChannel()]) { val1, val2 ->
/**
* The delegate inside Operators/Selectors is
* DataflowProcessor
*
*/
registerChannelListenersToAllInputs(ChanlListener)
println "***** The value is of d N e is: $val1 N $val2"
}
def selector1 = selector(inputs : [d.createReadChannel(), e.createReadChannel()]) {val, idx ->
println "%%%%%%% The SELECTOR Val is: $val \n Index: $idx \n"
}
/**
* Within Selectors we can specify outputs too just like normal Operators.
* For the sake of simplicity we have omitted that.
*
*/
/**
* End of Block for Broadcast
*/
/**
* Block for Queue
*
*/
/*def op1 = operator(inputs:[d, e]) { val1, val2 ->
println "***** The value is of d N e is: $val1 N $val2"
}
def selector1 = selector(inputs : [d, e]) {dVal, eVal ->
println "%%%%%%% The SELECTOR dVal is: $dVal \n eVal: $eVal \n"
}*/
/**
* End of Block for Queue
*/
/*def op2 = operator(inputs:[e]) { val ->
println "The value is of e: $val"
}*/
def op3 = operator(inputs:[highVolumeQueue]) { highVal ->
println "The High vOlume DATA: $highVal"
}
operator(inputs: [a, b, c], outputs: [d,e,highVolumeQueue], listeners: [listnrs]) {x, y, z ->
println "x:$x \n y:$y \n z:$z \n By Thread ${Thread.currentThread().isDaemon()}"
println "===== Binding First Output ====="
bindOutput 1, x+y+z
TimeUnit.MILLISECONDS.sleep(5000)
println "===== Binding Second Output ====="
bindOutput 0, x+z
if(x+2*y+z >= 80 )
bindOutput(2,fireCustomEvent(x+2*y+z))
else
bindAllOutputs x+2*y+z
}
task{
println "In Assignment Task.....By Thread ${Thread.currentThread()}"
a << 10
b << 20
TimeUnit.MILLISECONDS.sleep(6000)
c << 30
println "End Of Assignment Task.....By Thread ${Thread.currentThread()}"
}
task
{
println "Doing some Stupidity in the mean time..... :-) :-)"
}
[op1,op3]*.join()
/**
* Termination of the Operators/Selectors can be done with
* terminate(), terminateAfterNextRun()
* These methods call the Life Cycle Event Methods of the registered Life Cycle Event Handlers of the
* corresponding Selectors/Operators.
*
* Apart from these we can also use PoisonPill Control Message,
* which needs to be passed to the inputs of the Operators/Selectors
* i.e. Channels/Broadcasts/Variable etc and Event Methods of the registered Life Cycle Event Handlers
* gets invoked.
*
*/
}
catch(e)
{
e.printStackTrace()
}
println "End of Main........ By Thread ${Thread.currentThread()}"
}
}
class DataflowEventsListenerImpl implements DataflowEventListener {
public void registered(DataflowProcessor processor) {
// TODO Auto-generated method stub
println "******* Registered ...."
}
public void afterStart(DataflowProcessor processor) {
// TODO Auto-generated method stub
println "******* afterStart ...."
}
public void afterStop(DataflowProcessor processor) {
// TODO Auto-generated method stub
println "******* afterStop ...."
}
public boolean onException(DataflowProcessor processor, Throwable e) {
// TODO Auto-generated method stub
println "******* onException1 .... ${e.printStackTrace()}"
return false;
}
public Object messageArrived(DataflowProcessor processor,
DataflowReadChannel<Object> channel, int index, Object message) {
// TODO Auto-generated method stub
println "******* messageArrived .... $message for Channel $channel at Index $index"
return message;
}
public Object controlMessageArrived(DataflowProcessor processor,
DataflowReadChannel<Object> channel, int index, Object message) {
// TODO Auto-generated method stub
println "******* controlMessageArrived .... $message for Channel $channel at Index $index"
return message;
}
public Object messageSentOut(DataflowProcessor processor,
DataflowWriteChannel<Object> channel, int index, Object message) {
// TODO Auto-generated method stub
println "******* messageSentOut .... $message for Channel $channel at Index $index"
return message;
}
public List<Object> beforeRun(DataflowProcessor processor,
List<Object> messages) {
println "******* beforeRun .... messages $messages"
// TODO Auto-generated method stub
return messages;
}
public void afterRun(DataflowProcessor processor, List<Object> messages) {
// TODO Auto-generated method stub
println "******* afterRun .... messages $messages"
}
public Object customEvent(DataflowProcessor processor, Object data) {
// TODO Auto-generated method stub
println "******* customEvent .... for Data: $data"
return 'I am Firing Custom Event';
}
}
class DataflowChannelListenerImpl implements DataflowChannelListener
{
public void onMessage(Object message) {
// TODO Auto-generated method stub
println "@@@@@@@ Channel Listener invoked with Input"
}
}
In this code snippet we have introduced Listeners for Operators which responds to Opeartor's LifeCycle Phases and Events and another Listener for DataflowQueue.
The Callaback methods within Listeners which gets executed at the onset/end of Opeartor's various LifeCycle Phases and are itself self explanatory.
Unlike previous example, we have used DataflowBroadcast and DataflowQueue instead of DataflowVariable.
There is a task populating DataflowQueue a,b & c with some delay. As soon as they are populated an operator acts on them and binds them manually (output channels represented by index ordering) to different output channels based on some condition.
There are other Selectors and Operators on those output channels which acts accordingly. The main objective behind this example is to demonstrate the usage of Listeners.
HEY!!! I am back with Actors, one of the GPars interesting features.
I will start with Actor
Actor is such an Entity which allows only one thread at a time, irrepective of the object the thread belongs!!!!
Lets start with a simple exemplar.
import groovyx.gpars.actor.Actor
import groovyx.gpars.actor.Actors
import java.util.concurrent.TimeUnit
class ActorTest {
static def msg11="Default"
static def startTime;
/**
* Ac Actor is initiated as Actors.actor{...
*
*/
static def actor = Actors.actor{
println " Actor Initialized by:${Thread.currentThread().getName()}"
println "The metaclass is: ${delegate.getClass()}"
/**
* The Delegate inside Actors.actor Closure is Default Actor
*
*/
loop{
println "Inside Actor Loop :${Thread.currentThread().getName()}"
react(20000) {msg2 ->println "The Msg-3 is $msg2 :${Thread.currentThread().getName()}"
if (msg2 == Actor.TIMEOUT) {
//continue conversation
println "Thank you for $msg2 I am terminating Myself as Time Difference is ${new Date().getTime() - startTime}"
terminate()
}
msg2 = msg2+msg2
msg11 = msg2 + ' Subha '
//reply "Received!!!!"
//println "Sleeping Start.....The Current Thread is (Actor-33):${Thread.currentThread().getName()}"
Thread.sleep(8000)
msg11 += ' Paul '
//println "Sleeping End....."
}
println "Exiting Actor Loop :${Thread.currentThread().getName()}"
}
}
static mains (args)
{
startTime=new Date().getTime()
TimeUnit.MILLISECONDS.sleep(4000)
15.times {print "*"
}
println ''
println "The Main Thread.... Before Sending 1: ${Thread.currentThread().getName()}"
def a = actor.send "1"
println "The Main Thread.... Before Sending 11: ${Thread.currentThread().getName()}"
def b = actor.send "11"
15.times {print "*"
}
println ''
println "The Main Thread.... Before Sending 111: ${Thread.currentThread().getName()}"
TimeUnit.MILLISECONDS.sleep(15000)
def c = actor.send "111"
println "..... Sleep Finished...."
//Thread.sleep(10000)
[actor]*.join()
println "The msg is $msg11"
}
}
The main consideration behind the above mentioned code block is to exhibit that at a time only One thread can enter the critical area inside the Actor.
We have send messages to the Actor from the main thread thrice , but all these messages are being processed sequentially by the Actor in a Separate thread, and the main Thread thus quits, because all these Threads that execute the job inside the actor are Daemon threads, so in order to make the programme excute (as we have introduced delay here), we have joined the actor in the end as:
[actor]*.join()
This will make the Main Thread to waitfor the Actor(s) to finish, but as we have introduced loop inside the Actor, the actor will keep on running until we manually terminate the Actor, so in order to do that, we have used react() [N.B. Th reach() is the method which reacts or we can say responds to the messages being sent to the Actor] with a Timeout, which will make the actor wait for 20000 ms max for another message to arrive, and if there has been no message within that timeframe, then a Actor.TIMEOUT will be thrown and the Actor will terminate and thus terminating main.
We have also intrduced delay inside the Actor to show that, while a thread is sleeping, other threads, that has been started from the main thread, do not get the chance to execute, until and unless the current thread finishes its execution.
Actor has another propitious utility i.e. reply, which we have commented out here. Please check the GPars for furthur details.
We can execute the Tasks inside Actor by Non-Daemon Threads, by using Non-Daemon thread pool.
The Output of the above code will be:
Actor Initialized by:Actor Thread 1
The metaclass is: class groovyx.gpars.actor.DefaultActor
Inside Actor Loop :Actor Thread 1
Exiting Actor Loop :Actor Thread 1
***************
The Main Thread.... Before Sending 1: main
The Main Thread.... Before Sending 11: main
The Msg-3 is 1 :Actor Thread 1
***************
The Main Thread.... Before Sending 111: main
Inside Actor Loop :Actor Thread 1
Exiting Actor Loop :Actor Thread 1
The Msg-3 is 11 :Actor Thread 1
..... Sleep Finished....
Inside Actor Loop :Actor Thread 1
Exiting Actor Loop :Actor Thread 1
The Msg-3 is 111 :Actor Thread 1
Inside Actor Loop :Actor Thread 1
Exiting Actor Loop :Actor Thread 1
The Msg-3 is TIMEOUT :Actor Thread 1
Thank you for TIMEOUT I am terminating Myself as Time Difference is 48077
The msg is 111111 Subha Paul
The above outputs are being printed with some deplay, and which is interesting..
Lets track another Actor example
import groovyx.gpars.actor.DefaultActor;
import groovyx.gpars.actor.Actors;
class ActorTest2 {
class Act1 extends DefaultActor
{
def secretNum
void afterStart() {
println " ***** After Start is invoked ***** "
secretNum = new Random().nextInt(10)
println " ***** After Start is invoked done***** "
}
void act() {
println "**** Act is Invoked **** "
loop {
println "**** Loop is Invoked **** "
react { int num ->
println "**** Reaction Started for $num **** "
if(num == 10){
reply 'you win'
terminate()
}
else if (num > secretNum)
reply 'too large'
else if (num < secretNum)
reply 'too small'
else{
reply 'you win'
println "Terminating Myself....."
terminate()
println "Terminated....."
}
println "**** Reaction Finished for $num **** "
}
println "**** Loop is Finished **** "
}
}
}
class Act2 extends DefaultActor
{
String name = 'Subha'
Act1 server = new Act1()
int myNum
void afterStart() {
println " ***** After Start is invoked for Act2 ***** "
server.start()
println " ***** After Start is invoked done for Act2 ***** "
}
void act() {
loop {
myNum = new Random().nextInt(10)
server.send myNum
react {
println "The Sender is: $sender"
switch (it) {
case 'too large': println "$name: $myNum was too large"; break
case 'too small': println "$name: $myNum was too small"; break
case 'you win': println "$name: I won $myNum"; endActor.send server ; terminate(); break
}
}
}
}
}
static def endActor = Actors.actor{
//loop{
react{ server1 ->
println "%%%%%% End Actor is invoked %%%%%%"
server1.stop()
// terminate()
println "%%%%%% All Terminated %%%%%%"
}
//}
}
static def testActor = Actors.actor{
react { obj ->
println "The Object is $obj"
}
}
static def loopTestActor = Actors.actor{
def counter =1
loop{
println "Entering Loop.......... $counter"
react { obj ->
println "The Object is $obj"
}
println "Ending Loop.......... ${counter++}"
}
}
/**
* Usually the react method accepts a single Msg
* Now in order to make it accept more than one i/p
* parameter, we need to nest react
*
*/
static def calculator = Actors.actor{
react{ a ->
println "The First input is: $a Thread-${Thread.currentThread()}"
react{ b ->
println "The second input is: $b Thread-${Thread.currentThread()}"
console.send(a+b)
}
}
}
static def console = Actors.actor{
react{result ->
println "The result of addition is: $result Thread-${Thread.currentThread()}"
}
}
static main(args)
{
//def act1 = new ActorTest2.Act1(new ActorTest2())
//def act1 = ActorTest2.Act1.newInstance(new ActorTest2())
def act2 = ActorTest2.Act2.newInstance(new ActorTest2())
act2.start()
[endActor]*.join()
testActor << "Okkkiii----9"
testActor << "Nexttt-Okkk"
testActor.join()
20.times {
print "="
}
println ""
loopTestActor << "Okkkiii"
loopTestActor << "Nexttt-Okkk"
loopTestActor.join()
println "============End In Main Join............."
calculator.send 5
calculator.send 9
calculator.join()
}
}
As in the previous example we have initialized Actor with Actors.actor {...., here we have extended DefaultActor and created an Inner class for the Actor.
DefaultActor contains some Lifecycle methods Like:
1) afterStart() -> Gets invoked after the Actor is started explitly using the start() method.
2) act() -> This method reacts to the incoming messages.
The above example depicts different aspects of actors, but the significant one is, how an actor replies to the messages sent by another Actor.
In the above example there are Actors with names Act1, Act2 and endActor
In the main method we have initiated Act2 and started it and made the main Thread to wait until the endActor has finished execution
After we have started Act2, from within the afterStart() method we have started Act1.
Now the Act2 actor sends messages to Act1 actor and which reacts accordingly to the messages of Act2 by sending appropiate replies, which the Act2 processes and on termination endActor is invoked.
Another important point is if there is no loop inside an Actor then it can react to only the first message send to the actor. Besides , an Actor has a handful of built in variables like sender, which denotes the sender of messages and etc.
The topic Concurrency is and always will form one of the integral part of any programming language, be it Java, Groovy, Scala etc.
What I have always felt, that this thing is one of the key factors in determining how powerful the programming language is and at the same time whenever we come across this term, the thing which comes to our mind is:
Yes!!!! To me, the topic Concurrency be it in any language, at first it becomes difficult for me to get the ideas behind all the concepts, but as I keep on learning and keep on solving puzzles then things turns out to be more and more intersting and engaging.
So I always consider Concurrency to be a rivetting theme.
Now, the Concurrency in Java using Thread model which we all have experienced more or less, differs well with the natural human sense for parallelism.
Now Why Concurrency is becoming an integral part of any computing that we do?
Beacuse due to advent and fast propagation of MULTICORE CPU's, which is almost a commonplace routine nowadays.
So It is now high time that we should immediately embrace and ratify those means which will not respect the sense of parallelism that we create in our brain but also mental models of Concurrent systems interactions.
Propitiously Groovy replenish such awesome features.
The Concurrency traits that Groovy comes up with are as follows:-
I tried to explore some of these splendid concepts and I am really Impressed!!!!!. Thanks to GPars and Groovy.
I will share my experience with some code blocks which I have followed to find out how these elements work.
Keep a close WATCH on this space for something really interesting......
1) Inspect the Request Body at the Filter (as sent by the POST method) .
2) Setting a new Request Body depending on some Business Rules, i.e. either altering the existing request body and setting a new request body for the following filters or Resource Actions.
3) Depending on some error condition, to stop the request from travelling furthur i.e. into following filters or ResourceAction and setting the response body.
Before elaborating on all the above mentioned points let me recapitulate some of key gServ Control Flow points, which I already have described in my previous blogs CONTROL FLOW that:
When an incoming request arrives, the GServHandler's handle(HttpExchange) method gets called.
This method mainly does the following things:
1) Creating of the RequestContext from already available GServConfig with the help of
new GServFactory().createRequestContext
2) Finding all the registered filters from the GServConfig that matches the Request Action.
3) Executing Filters with the help of FilterRunner
4) And if the processing for the Filters is fine then forwarding the request to ResourceAction for processing else closing the RequestContext after writing the response to the outputstream.
The code block for the filter is:
def beforeFilter = ResourceActionFactory.createBeforeFilter "bookBeforeFilter3", "POST", "/data/book", [:], 5, {requestContext, data ->
log.info " ********** I am Done!!!!!! *********** for requestContext: ${requestContext.getClass()} and Data ${data.getClass()} \n\n Data MetaClass ${data.metaClass}"
def offset = 0
def error = true
def byteArr = new byte[1024]
log.info "**** Size: ${data.size()}"
InputStream inStream
def byteArrayOutputStream = new ByteArrayOutputStream();
def result = '{"id":"11", "name":"GROOVY IN ACTION 2nd EDITION"}'
byteArrayOutputStream.write(result.getBytes())
byteArrayOutputStream.close()
inStream = requestContext.getRequestBody()
//IOUtils.copy(inStream,byteArrayOutputStream)
def byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray())
def errorStream = new ByteArrayOutputStream()
errorStream.write("There is an error so Redirecting.....".getBytes())
errorStream.close()
def count = inStream.read(byteArr, offset, byteArr.length)
while(count > 0)
{
log.info "%%%%%% The Data Read Count is $count"
offset += count
log.info new String(byteArr)
count = inStream.read byteArr, offset, byteArr.length
}
inStream.close();
if(error)
{
/**
* In case of any error we are not forwarding the
* request to any following filters or Resource Action,
* Rather we can set the Response Body or Response Headers
* and at the end close the RequestContext
*/
log.info "=========== Since error was detected so closing the RequestContext $requestContext ============"
/**
* The Below lines really describes the Magic of Groovy...
* I'm imprrrrressed....
*
*/
requestContext._responseBody = errorStream
requestContext.close()
}
else
requestContext.setRequestBody byteArrayInputStream
requestContext
}
Inspect the Request Body at the Filter (as sent by the POST method)
Now from the above code we can see that the parameters to the closure of the filters (which is the Request Handler for this After Filter) are:
1) requestContext (Type is RequestContext and the actual implementation being RequestContextWrapper)
2) data (Type is List and implementation being ArrayList)
Now we can get the Request data from both of them, for the first option with the help of getRequestBody() which we have done here, or from the data which is an ArrayList, from the first element of this ArrayLisy which is of Type InputStream(FixedLengthInputStream), but the catch is once we read the InputStream with any of the practices as described above, we cannot read it again (as java.io.IOException: Stream is closed), so after reading if we close the inputstream and forward the request, then the next entities in the control flow will not be able to fetch the data from the InputStream.
so we did the following:
def byteArrayOutputStream = new ByteArrayOutputStream();
def result = '{"id":"11", "name":"GROOVY IN ACTION 2nd EDITION"}'
byteArrayOutputStream.write(result.getBytes())
byteArrayOutputStream.close()
.........
........
def byteArrayInputStream = new ByteArrayInputStream(byteArrayOutputStream.toByteArray())
................
.................
requestContext.setRequestBody byteArrayInputStream
Here we have created a new Input Stream and set it within the Request Body.
The second point i.e. Setting a new Request Body is also being covered here.
Stopping the request from travelling furthur
For this we have created an error flag, and an error InputStream and then we have closed the current RequestContext after the setting the desired response within ResponseBody
def error = true
..........
.........
def errorStream = new ByteArrayOutputStream()
errorStream.write("There is an error so Redirecting.....".getBytes())
errorStream.close()
......................
......................
if(error)
{
/**
* In case of any error we are not forwarding the
* request to any following filters or Resource Action,
* Rather we can set the Response Body or Response Headers
* and at the end close the RequestContext
*/
log.info "=========== Since error was detected so closing the RequestContext $requestContext ============"
/**
* The Below lines really describes the Magic of Groovy...
* I'm imprrrrressed....
*
*/
requestContext._responseBody = errorStream
requestContext.close()
}
We have also introduced a Custom Converter within :
/**
* Custom Converters - To convert Request Body (JSON format here)
* to the expected Object Type
*
*/
def server = gserv.plugins{
plugin ( 'compression', [:] )
plugin ('cors',[:])
}.http {
........
........
def converter = conversion(Book.class){ instream ->
def data = new JsonSlurper().parse(instream)
new Book(name: data.name,
id: data.id,
author: data.author,
price: data.price,
sellers:data.sellers,
authorAddress:data.authorAddress,
repo:data.repo)
}
}
This Custom converter converts the input JSON into Book type as expected by:
post "/book", onlyIfAccepts('text/json'), /*onlyIfHeader('Accept-Encoding',"gzip"),*/ {Book book2 ->
log.info "!!!!!!!! The Resource Action Delegate is ${delegate.getClass()}"
log.info " The POSTED data is:$book2"
log.info "Sleeping Thread....."
//sleep(20000);
def onlineRepo = new OnlineRepo(id:"GITHUB",repoURL:"http://github.com/gServ")
def address1 = new Address(building:"CHURNI",street:"PAS",city:"Kol",country:"India",zip:222334)
def address2 = new Address(building:"CHURNI2",street:"PAS2",city:"Kol2",country:"India2",zip:222334)
def address3 = new Address(building:"CHURNI3",street:"PAS3",city:"Kol3",country:"India2",zip:555334)
def book3 = new Book(name:"Groovy In Action 2nd Edition",id:"11",author:["Gordon","Dickens"],price:12.34,sellers:["seller1":"pearson","seller2":"Tata Mc Graw"],authorAddress:[address1,address2,address3],repo:onlineRepo)
def jsonOutput = JsonOutput.toJson(book3)
log.info "The JSON OUPTPUT is :$jsonOutput"
log.info "Waking Up and Pocessing......"
//book = new Book(name:"Groovy REST IN Action",id:'ISBN:9001',author:'Gordon Dickens',price:120.89)
responseHeader "content-type", "application/json"
write book2.toString().bytes
}
So this is in short about filters in gServ.
From the all our discussions and analysis on GServ we can say, that for each and every element in this framework, be it Resource, ResourceAction, Filters, ServerInstance, there is a delegate which does all the configuration and so that GServconfig can be populated and when the request arrives, then based on URL, Method, Parameter matching we execute those elements with the help of some Runner elements like for Filter there is FilterRunner, for ResourceAction there is ActionRunnner etc.
Please share your valuable comments and Suggestions.
In this post we will see how the GServConfig i.e. the config object gets constructed, followed by Control flow for Incoming Requests.
** GServInstance.start(<<PORT NUMBER>>). This method creates the HttpServer and registers the GServHandler (created in Step 3.1) with the context root "/". The GServHandler which extends HttpHandler acts like controller after it gets registered.
Along with this some init Filters also gets registered and stored in GServConfig, so that when the filters gets invoked from GServHandler.handle(HttpExchange) method, then these init filters also gets executed.
Remember the context root being "/".
So for any incoming requests, the HttpHandler’s handle(HttpExchnage) gets invoked.
The HttpExchange class encapsulates a HTTP request received and a response to be generated in one exchange. It provides methods for examining the request from the client, and for building and sending the response.
So when the incoming request arrives the HttpExchange instance along with the GServConfig object (passed in Step 3) to create the GServ RequestContext, which is then passed along for invoking filters, plugins, Actions for a particular Request.
Please note, the depicted control is an overview of my understanding of gServ.
There can be Gap between my understanding and the actual flow. Please share your views and feedback.
If I am wrong or misunderstood some part of the gServ architecture and the control flow, I apologize for that and please correct me in that case.
I thought of sharing my idea about gServ because I thought that such effort may help us in knowing this awesome Groovy framework in a better way.
That’s it and until next time Keep Coding and Keep Contributing.
A short while ago while I was exploring Spring HATEOAS , which is gradually becoming one of my dearest Spring modules and which I will surely explore furthur, I was thinking whether Groovy provides any such Rest framework .
And the answer I found is "YES" and it is known as gServ .
Though there is not too much documentation for it at present what whatever I can find from GITHUB along with my own findings, I will share it with you.
Before I start I must say that the GITHUB really helped me a lot to understand gServ features and all its setup related activities.
Lets start Now!!!
So the first question that comes to our mind that why gServ. We have such a robust Spring Rest features along with Spring HATEOAS module, what is more there in gServ that will interest us.
The gServ traits are as follows:
1) It is container free. Yes you heard it right!!!! the Rest Services can be deployed using HttpServer.
2) It serves static files.
3) It provides Plugins and HATEOAS support.
4) It has got support for CORS (Cross Origin Resource Sharing).
5) Provides Compression and ETag support and others.
Along with all these above mentioned features, we also need to get acquainted with some terms, specific to gServ before we proceed furthur.
Like:
1) gServ Resource.
2) gServ Resource Action.
3) gServ Server Instance.
4) gServ Server Config etc.
Lets Describe all of these with referenes to Code.
Below is a small piece of code block containing basic gServ features.
@SpringBootApplication
@Slf4j
class RestApp {
static main(args)
{
def book
def gserv = new GServ()
def bookResource = gserv.resource("/data"){
log.info "!!!!!!!! The Resource Delegate is ${delegate.getClass()}"
def resourceAction = post "/book", onlyIfAccepts('text/json'), /*onlyIfHeader('Accept-Encoding',"gzip"),*/ {
log.info "!!!!!!!! The Resource Action Delegate is ${delegate.getClass()}"
book = new Book(name:"Groovy REST IN Action",id:'ISBN:9001',author:'Gordon Dickens',price:120.89)
responseHeader "content-type", "application/json"
write book.toString().bytes
}
log.info "@@@@@@@@@@@@@ The Resource Action is ${resourceAction.getClass()}"
get "/bookNoEnc", onlyIfAccepts('text/json'), onlyIfHeader('Accept-Encoding',"gzip") , {
log.info "The Request Context class is1111: ${requestContext.class}"
book = new Book(name:"Groovy REST IN Action 2nd Ed",id:'ISBN:9001',author:'Gordon Dickens',price:125.89)
write book.toString().bytes
}
}
log.info "##### The BookResource is $bookResource #####"
def beforeFilter = ResourceActionFactory.createBeforeFilter "bookBeforeFilter3", "POST", "/data/book", [:], 5, {requestContext, data ->
log.info " ********** I am Done!!!!!! *********** for requestContext: ${requestContext.getClass()} and Data $data"
null
}
def server = gserv.plugins{
plugin ( 'compression', [:] )
plugin ('cors',[:])
}.http {
/*weakETag("/data/bookNoEnc"){ requestContext, byte[] data ->
Encoder.md5WithBase64(args.join('//').bytes)
}*/
log.info "!!!!!!!! The Http Server Delegate is ${delegate.getClass()}"
before "bookBeforeFilter2", "/data/book", "GET", [:], 4, {requestContext, data -> log.info "*********** ${requestContext.class} \n The data is $data"}
before "bookBeforeFilter1", "/data/book", "GET", [:], 4, {requestContext, data ->
log.info " ********** I am Done..... *********** "
requestContext
}
log.info "***** ${bookResource.basePath}"
log.info "***** ${bookResource.linkBuilder}"
addFilter beforeFilter
resource bookResource
}.start(9000)
log.info "The Server is: ${server.class}"
}
}
The above mentioned code block do not provide comprehensive coverage of the gServ features but will help us to get some idea regarding how gServ acts and all the contituent participants that makes gServ interesting.
1) gServ Resource
At the beginning of the above code block we can see that we have instantiated a class GServ. This is heart and soul of the gServ framework.
when invoking resource() on the GServ instance by passing the root path and the closure we create the gServ Resource.
Within Closure we pass the sub-elements which constitute the gServ Resource Actions.
The delegate of the closure is ResourceDelegate. This class with the help of the inherited behaviour of the trait (i.e. ResourceFn)
provides support.
A Resource in gServ can be considered as an entity on which we want to perform actions, like fetching info (GET), altering (PUT), Creating new Resource (POST), deleting a resource (DELETE) etc. Each gServ resource is associated with a basepath which is here "/data"
2) gServ Resource Action
One of the element as passed in the GServ.resource() method's closure is:
post "/book", onlyIfAccepts('text/json'), /*onlyIfHeader('Accept-Encoding',"gzip"),*/ {
log.info "!!!!!!!! The Resource Action Delegate is ${delegate.getClass()}"
book = new Book(name:"Groovy REST IN Action",id:'ISBN:9001',author:'Gordon Dickens',price:120.89)
responseHeader "content-type", "application/json"
write book.toString().bytes
}
This is the Resource action.
A Resource action is the Request Handler for a particular path/query/method combination, i.e. this resource action will handle requests having HTTP method POST, for the path "/data/book" (prepending the root/base path), and with the Request Header "Accepts" having value 'text/json'
The Delegate to the closure passed to the Resource Action is HttpMethodDelegate.
Each Resource Action can refer to the RequestContext from within the closure. The RequestContext is the object which is created by GServHandler when a request arrives for a particular endpoint for a particular path/query/method combination and which contains the request related details and we can also set response headers or fetch related info from this object. It has lot of other utilities the details can be checked out from the official documentation. It is really great out there.
So within the Request Action Closure we define the processing logic corresponding to the path/query/method combination and along with that decide on the response format (In the above example we have simply returned the dummy response as bytes) and set response headers if required and do other stiffs.
In the above code Block we have also defined some BEFORE filter for a particular path/query/method combination.
def beforeFilter = ResourceActionFactory.createBeforeFilter "bookBeforeFilter3", "POST", "/data/book", [:], 5, {requestContext, data ->
log.info " ********** I am Done!!!!!! *********** for requestContext: ${requestContext.getClass()} and Data $data"
null
}
Now that we have Resource, Resource Action, Filters in place we need to bring all of these components into Action. But the question arisis who will integrate all these components together and serve the requests.
gServ Server Instance.
The GServInstance comes to the rescue.
We create that instance with the help of GServ.http() method and then at last we start the instance(i.e. the http server) with the help of start() method ( within start() method we pass in the port number here 9000, where the http server will be listening..... ), which eventually brings all the resources along with their resource actions, filter and other components in action.
gserv.http{
}.start(9000)
The delegate to the closure passed in to the http() is the ServerInstanceDelegate.
So whatever method call we define in the closure gets invoked from the Delegate.
Now that we know who inegrates all the components and makes them work together, The next question that arises is how does all these things takes place from registering resources, plugins, filters etc, to serving the requests when they arrive at a particular endpoint????
Lets describe the registration part first:
When we declare the following within the closure:
.http{
before "bookBeforeFilter2", "/data/book", "GET", [:], 4, {requestContext, data -> log.info "*********** ${requestContext.class} \n The data is $data"}
before "bookBeforeFilter1", "/data/book", "GET", [:], 4, {requestContext, data ->
log.info " ********** I am Done..... *********** "
requestContext
}
log.info "***** ${bookResource.basePath}"
log.info "***** ${bookResource.linkBuilder}"
addFilter beforeFilter
resource bookResource
}
The elements gets registered by the ServerInstanceDelegate within a Map declared in DelegateFunctions.
gServ Server Config
The following is a part of Server Config (i.e.GServConfig.java), we will get a more clear picture when we will explore the request flow from GServ :-).
Each key of the Map is the type of elements and has an associated List as the value for the Key, so that elements gets registered within that List, what I mean by this is for filters there is a key within the map named "filterList" and this key has a List associated with it, so all the filters gets registered within that list.
Likewise for resources there is "actionList" which contains all the actions corresponding to that resource.
In our case there are two actions for the BookResource, one is GET for the action path "/data/bookNoEnc" and other is POST for action path "/data/book".
One more thing, one significant task that http() method does apart from registering the above mentioned artifacts, is creating the GServConfig instance which contains all the configuration related data which we will be using while handling the incoming requests.
So in very brief this is how the Registration of the constituent elements does happen.
Now the question arises how the control flows, when the request arrives.
That I have tried to describe with a flow diagram Here. Please share your views and feedback and Till next time keep coding and keep contributing.
Generics is one of the most elemental concepts of Core Java.
Whenever we talk about Generics, the leading thoughts that sprout in our mind are Type Parameter, Type Erasure etc.
To be honest, I always find Generics concepts a little hard to grasp at the very first go, but once we start adapting it, it is becomes more and more interesting and very useful.
This post is not intended to deliver any tutorial, or any introduction to basics of Generics, but about scenario, which I confronted recently and I think I should bring it into light.
We all more or less at some point in our coding lifetime, have encountered Generics, especially while using Collection Framework elements (Maps, Lists, Sets etc.)
With all these elements we basically use the instance methods of the Objects, but what about when invoking static Methods.......
Lets describe it with an example:
public class XYZ {
T t1;
T t2;
public void build()
{
System.out.println(" ****** In Build ******* "+((t1 != null)?t1.toString():"null")+"-----"+(t2 != null?t2.toString():"null"));
}
}
The class XYZ has a Type Parameter T and we have declared two references of Type T and have accessed them in a instance method.
Upto this point the compiler is very Happy and doesn't seems to be complaining.
But what about we introduce a static method and try to access the references of type T within it,
Now, the compiler will now exhibit its dislike and the error we will get is Cannot make a static reference to the non-static field ...., which is obvious..
So it becomes:
public static XYZBuilder buildStatic(Class m)
{
System.out.println(" ****** In BuildStatic"+t1);
return new XYZBuilder(m);
}
Oooops!!!!! Sorry, I forgot to introduce another member XYZBuilder, I will don't worry.....
So to get rid of the error, the first thing that we will do is:
static T t1;
and I too did the same thing, but will my surprise, I found that it did not worked, and instead I accompanied myself with the same error, which is: Cannot make a static reference to the non-static field ....
And after analysis I found this to be absolutely logical, as the Type parameter are belongings of instances of class XYZ, whenever we create an instance of the class , Type parameter will be bound to that object so cannot be conferred a static context, as making it static means the Type Parameter belongs to the Class, irrespective of the objects created, which is absolutely not!!!!!!
So I found the reason, but the actual problem still persists, as how to use a Type Parameter within a static Context and the Answer is just declare the Type Parameter Here "M" between the static modifier and the return type of the static method and we are free to use it within the static context, like
public static XYZBuilder buildStatic(Class m)
{
System.out.println(" ****** In BuildStatic");
return new XYZBuilder(m);
}
We have used the wildcard character ? within the method for flexibility so that we can pass Class of any Type, but we can also restrict this with Class or Class
Now how to call the static Method????!!!!!
Seems pretty Ordinary Right.....what is so particulat in calling a static method of a class.
The answer is:
XYZ.buildStatic(Object.class);
but this is the call using Raw, where is the Type Parameter ???, that we have declared in the static Context,
and the Answer is:
The part to look at is the XYZ.<String>buildStatic(Object.class);
Great isn't it!!!!!
Now its time to introduce all of you to the another awaited member of the family the XYZBuilder
Its definition is:
public class XYZBuilder {
T payload;
Class t2;
XYZBuilder()
{
}
/*XYZBuilder(Class t)
{
this.t2 = t;
}*/
XYZBuilder(Class t)
{
this.t2 = t;
}
public XYZBuilder payload(T payload)
{
this.payload = payload;
System.out.println("The Payload type is:"+payload.getClass());
return this;
}
public XYZ build()
{
return new XYZ();
}
}
With this XYZBuilder class, we will be able to exhibit the utility of the Type Parameter "M" introduced within the static context.
From buildStatic() method we are returning an instance of XYZBuilder, with a Type Parameter, this Type Parameter is the type of payload, in XYZBuilder class, so whatever Type we pass in the static context, that becomes the integral part of XYZBuilder class, and we have captured the Class type thouugh constructor in another variable for Record purpose only.
The Test class looks alike:
public class Test2 {
public static void main(String[] args) {
// TODO Auto-generated method stub
//new XYZ().build();
XYZBuilder xyzBuilder= XYZ.buildStatic(Object.class);
System.out.println(xyzBuilder.t2);
xyzBuilder.payload("okkkk");
//XYZ.buildStatic(Object.class);
}
}
The output will be:
****** In BuildStatic
class java.lang.Object
The Payload type is:class java.lang.String
So Generics is really intriguing.
Please share your views or any other GENERICS feature which you consider to be appealing.
In this concluding episode of "My Tryst with Groovy DSL" we are going to stress on building various components with Groovy DSL.
Lets say,
We have an annotation Imitate as:
So, to put it in words, our objective is:
To declare two closures (mi3,mi4) one having static scope.
Then we will declare a new method NEWOKNew1() and invoke the closures in order and all these operation we going to implement with Groovy AST using transformation class.
The GroovyASTTransformationClass class is:
@GroovyASTTransformation(phase = CompilePhase.CANONICALIZATION)
class ImitateTransformation implements ASTTransformation {
@Override
public void visit(ASTNode[] astNodes, SourceUnit sourceUnit) {
// TODO Auto-generated method stub
def statement = astNodes[1].code
def varscope = statement.variableScope
Map referencedLocalVariablesMap = varscope.referencedLocalVariables
referencedLocalVariablesMap.each { key , value ->
println "-->${key.class}"
println "-->${value.class}"
}
/**
*
* Source Unit is the Groovy Source Code File
* containing the annotation.
*
*/
println "===========The SourceUnit is ${sourceUnit.name}"
def shareVariables = { variableScope ->
println "The variable scope is: ${variableScope.referencedClassVariables}"
//def scope = variableScope.copy()
def map = variableScope.referencedLocalVariables
println "&&&& The map is:$map"
map.each{ key, value ->
println "***** The Variable is $value.name"
value.setClosureSharedVariable(true)
}
variableScope
}
shareVariables(statement.variableScope)
// create new block statement
/* BlockStatement block = new BlockStatement()
block.variableScope = shareVariables(statement.variableScope)
// create closure expression, use code as an argument
ClosureExpression closure = new ClosureExpression(Parameter.EMPTY_ARRAY, statement)
closure.variableScope = statement.variableScope.copy()
// create method call expression, use the closure as an argument
MethodCallExpression methodCall = new MethodCallExpression(new VariableExpression('this'),
'NewOk', new ArgumentListExpression(closure))
// add method call to the block statement
block.addStatement(new ExpressionStatement(methodCall))
nodes[1].code = block
*/
MethodNode annotatedMethod = astNodes[1]
ClassNode declaringClass = astNodes[1].declaringClass
def closre = new AstBuilder().buildFromSpec {
closure {
parameters {
parameter 'args2': String.class
}
block {
expression {
methodCall {
variable "this"
constant "println"
argumentList {
constant "Closure Call Done"
//parameter "args2"
}
//constant "OKKK"
}
}
expression {
methodCall {
variable "this"
constant "println"
argumentList {
//constant "Closure Call Done"
variable "args2"
}
}
}
}
}
}
closre[0].variableScope = new VariableScope()
/**
*
* @author MIC
*
* Defining another Closure
*
*/
def closre2 = new AstBuilder().buildFromSpec {
closure {
parameters {
parameter 'args21': String.class
}
block {
expression {
methodCall {
variable "this"
constant "println"
argumentList {
constant "##### Closure Declaration ######"
}
}
}
/*expression {
methodCall {
variable "this"
constant "println"
argumentList {
//constant "Closure Call Done"
variable "args"
}
}
}*/
}
}
}
closre2[0].variableScope = new VariableScope();//shareVariables(astNodes[1].getVariableScope())
/**
* Defining a static Closure
*/
def closre3 = new AstBuilder().buildFromSpec {
closure {
parameters {
parameter 'args31': String.class
}
block {
expression {
methodCall {
variable "this"
constant "println"
argumentList {
variable "args33"
}
}
}
expression {
methodCall {
variable "this"
constant "println"
argumentList {
variable "args31"
}
}
}
}
}
}
closre3[0].variableScope = new VariableScope()
//End of Closure Declaration
/**
* @author MIC
*
* Adding a field node to a class
*
*/
FieldNode fieldNode= new FieldNode('mi3', 1, new ClassNode(Object.class), new ClassNode(declaringClass.getClass()), closre2[0])
FieldNode staticFieldNode= new FieldNode('mi4', ACC_PUBLIC | ACC_STATIC, new ClassNode(Object.class), new ClassNode(declaringClass.getClass()), closre3[0])
//End of Adding a FieldNode
def ast = new AstBuilder().buildFromSpec {
method('NEWOKNew1', ACC_PUBLIC , Void.TYPE) {
parameters {
parameter 'args': String.class
}
exceptions{
}
}
}
def consCallExpr = new ConstructorCallExpression(new ClassNode(MyClass.class), MethodCallExpression.NO_ARGUMENTS);
def methCallExpr = new MethodCallExpression(new VariableExpression("mi4"), "setDelegate", consCallExpr);
PropertyExpression propertyExpression = new PropertyExpression(new ClassExpression(new ClassNode(Closure.class)), "DELEGATE_FIRST")
def methCallExpr2 = new MethodCallExpression(new VariableExpression("mi4"), "setResolveStrategy", propertyExpression);
ArgumentListExpression argListExpr = new ArgumentListExpression()
argListExpr.addExpression(new VariableExpression("args"))
argListExpr.addExpression(closre[0])
ArgumentListExpression argListExpr2 = new ArgumentListExpression()
argListExpr2.addExpression(new VariableExpression("args"))
MethodCallExpression methodCall = new MethodCallExpression(new VariableExpression('this'),
'testMeth',argListExpr)
MethodCallExpression methodCall2 = new MethodCallExpression(new VariableExpression('this'),
'mi3', new ArgumentListExpression())
StaticMethodCallExpression methodCall3 = new StaticMethodCallExpression(new ClassNode(MainExample.class), 'mi4', argListExpr2)
def methNode = ast[0]
def blockStatementList = [new ExpressionStatement(methodCall), new ExpressionStatement(methodCall2), new ExpressionStatement(methCallExpr2) ,new ExpressionStatement(methCallExpr), new ExpressionStatement(methodCall3)]
BlockStatement blockStatement = new BlockStatement(blockStatementList, new VariableScope())
methNode.setCode(blockStatement)
declaringClass.addProperty(new PropertyNode(fieldNode, ACC_PUBLIC, null, null));
declaringClass.addProperty(new PropertyNode(staticFieldNode, ACC_PUBLIC, null, null));
declaringClass.addMethod(methNode)
}
}
Now lets explain the the groovy transformation code :
At the beginning lies my some of findings with VariableScope, but really I was not able to figure it out how it is applicable and its true scope, It would be a great help, if someone could point me so, or help me in getting hold some resources, which will clarify my doubts :-) :-)
The there comes the closure declarations, closre, closre2 & closre3 and we also set variablescope for each of them, otherwise the AST changes won't be applied.
Like:
Then we decalre two properties of type FieldNode, and they point to closre2 & closre3 respectively.
FieldNode fieldNode= new FieldNode('mi3', 1, new ClassNode(Object.class), new ClassNode(declaringClass.getClass()), closre2[0])
FieldNode staticFieldNode= new FieldNode('mi4', ACC_PUBLIC | ACC_STATIC, new ClassNode(Object.class), new ClassNode(declaringClass.getClass()), closre3[0])
followed by setting Delegate & ResolveStrategy for closure pointed by mi4.
def consCallExpr = new ConstructorCallExpression(new ClassNode(MyClass.class), MethodCallExpression.NO_ARGUMENTS);
def methCallExpr = new MethodCallExpression(new VariableExpression("mi4"), "setDelegate", consCallExpr);
PropertyExpression propertyExpression = new PropertyExpression(new ClassExpression(new ClassNode(Closure.class)), "DELEGATE_FIRST")
def methCallExpr2 = new MethodCallExpression(new VariableExpression("mi4"), "setResolveStrategy", propertyExpression);
Then we declare the Argument List to be passed to testMeth(), mi3(), & mi4() respectively and we also defined the MethodCall expressions for testMeth(), mi3(), & mi4() with the help of Argument List prepared earlier.
ArgumentListExpression argListExpr = new ArgumentListExpression()
argListExpr.addExpression(new VariableExpression("args"))
argListExpr.addExpression(closre[0])
ArgumentListExpression argListExpr2 = new ArgumentListExpression()
argListExpr2.addExpression(new VariableExpression("args"))
MethodCallExpression methodCall = new MethodCallExpression(new VariableExpression('this'),
'testMeth',argListExpr)
MethodCallExpression methodCall2 = new MethodCallExpression(new VariableExpression('this'),
'mi3', new ArgumentListExpression())
StaticMethodCallExpression methodCall3 = new StaticMethodCallExpression(new ClassNode(MainExample.class), 'mi4', argListExpr2)
Lastly, we put all those MethodCall Expressions and PropertyExpressions in BlockStatement with a new VariableScope and it to the newly created method NEWOKNew1()
def methNode = ast[0]
def blockStatementList = [new ExpressionStatement(methodCall), new ExpressionStatement(methodCall2), new ExpressionStatement(methCallExpr2) ,new ExpressionStatement(methCallExpr), new ExpressionStatement(methodCall3)]
BlockStatement blockStatement = new BlockStatement(blockStatementList, new VariableScope())
declaringClass.addProperty(new PropertyNode(fieldNode, ACC_PUBLIC, null, null));
declaringClass.addProperty(new PropertyNode(staticFieldNode, ACC_PUBLIC, null, null));
declaringClass.addMethod(methNode)
methNode.setCode(blockStatement)
So, for this time thats all that I had been doing with Groovy AST.
Please feel free to share your views and comments and till next time keep coding and keep sharing.
For Mocking previously we have used, Mockito with Powermock, which extensively provides all the ways to deal with different scenarios (as posted HERE).
Few days back I came across a situation, where I need to perform Partial Mocking of a Service Class, I thought of exploring Mocking with Spock for something new, and now I can say that I am overwhelmed.
Let me brief about the scenario before going into the coding.
There is a service class, which has got a utility bean injected into it like:
@Service
class AppImpl implements App
{
public void main(File args) throws Exception {
// TODO Auto-generated method stub
println " I am Implementing $args"
}
public void subMain(String args) throws Exception {
// TODO Auto-generated method stub
println " I AM A SUBMAIN METHOD IN $args "
}
}
interface App {
void main(File args) throws Exception
void subMain(String args) throws Exception
}
So, FileProcessorImpl is the service class, and App is the utility bean type injected into the service class.
Now the requirement is, we want to test the functionality of FileProcessorImpl i.e. process(), but we want to mock the behaviour of App implementation called from within the Service, so that the actual implementation do not get invoked.
Now in order to to this we came up with 2 approaches:
a) Using interfaces b) Using Spock Mock/Spy
Using interfaces
We have written Test cases here using Spock framework:
Here is the Test class:
@ContextConfiguration(classes=EIPConfig)
class EIPConfigTest extends Specification
{
@Autowired
IProcessor fileProcessorImpl
def "Test Beans"()
{
given:
def output
def file = new File("C:/Users/MIC/Desktop/web.xml")
App appMock = [ main: {println ' ****** I am being Overridden ****** '}, subMain: {println ' ****** Submain being Overridden ******'}] as App
fileProcessorImpl.appImpl = appMock
when:
output = fileProcessorImpl.process(file)
then:
output==file
}
}
The annotation @ContextConfiguration defines a normal Spring Confuguration class which contains all the Bean definitions and also does Package scanning of the beans with @ComponentScan and registers all of them in the ApplicationContext.
So, due to @ContextConfiguration the IProcessor type is injected into the Unit Test Case.
Now if we have look into the given block, we can see that we have provided an implementation of the App interface, and then we have set this reference within the injected IProcessor type.
So when we are calling processorMock.process(.... then the mocked version will get called rather than the original.
Using Spock Spy
@ContextConfiguration(classes=EIPConfig)
class EIPConfigTest extends Specification
{
@Autowired
IProcessor fileProcessorImpl
def "Test Beans"()
{
given:
def output
def file = new File("C:/Users/MIC/Desktop/web.xml")
App mockApp = Spy(AppImpl)
mockApp.main(_) >> {println "****** I am being Overridden ****** '"}
//mockApp.subMain(_) >> { println ' ****** Submain being Overridden IN $it'}
fileProcessorImpl.appImpl = mockApp
when:
output = fileProcessorImpl.process(file)
then:
output==file
}
}
Now if we look into the above code snippet, then we can see that we have SPIED the Utility class i.e. AppImpl, and we have provided implementation to the main() only, while the submMain() that will be invoked will be from the Original implementation.
[Please Note: The underscore (_) represents any parameter, i.e. it is wildcard representation]
So, the output will be:
****** I am being Overridden ****** '
I AM A SUBMAIN METHOD IN C:\Users\MIC\Desktop\web.xml
Now if we uncomment subMain(), then the mocked implementation will be called, i.e. we will be providing implementation of only those methods which we want to mock, otherwise not, and then in that case the original implementation will be invoked.
Really this feature is helpful. Apart from using Spy(), you can also check out the Mock() or Stub() feature.
So till next time Keep contributing and Keep coding.
In my tryst with Groovy DSL, I went through some blogs describing groovy DSL features.
To explore groovy DSL features in a more vivid way I realized that I need to gather more insight on how groovy AST works.
In this blog I am going to add components to an existing groovy class using Groovy AST, I can co-relate (to some extent) that this groovy AST is somewhat like byte code engineering in Java, it changes the behaviour of components.
To do that I had an encounter with Groovy ASTBuilder class. Now this class does all that magic with the of its 3 methods:
With the help of the above 3 methods of the Groovy ASTBuilder class, we can change the Abstract Syntax Tree (AST) of a Groovy class.
Out of these 3 methods, I chose buildFromSpec as this presents a more Groovy DSL rich feature.
While I was browsing, to gain more on Groovy AST with ASTBuilder's buildFromSpec, I came across a post,which converts an annotated method into a main method, i.e. the main method do have all the code block of the annotated method.
I thought of why not integrate the main method in a class with some fresh code block, rather than taking the code block from the existing method,as this will enable to explore a little more on buildFromSpec.
Here the main method that I have implemented using ASTBuilder's buildFromSpec is a overloaded one, i.e. it contains two parameters of type String instead of one.
So here is the code:
The annotation that we will be using is @Main
@Retention (RetentionPolicy.SOURCE)
@Target(ElementType.TYPE)
@GroovyASTTransformationClass("com.groovy.ast.transformers.MainTransformation")
@interface Main {
}
For this annotation, the Retention type is Source, we will be using the annotation on Element Type and the Groovy AST Transformation class is MainTransformation, this is the class which does the byte code engineering on the class which we will annotate with @Main
Now lets describe things a little:
When we annotate the class MainExample with @Main, the Groovy Transformation class MainTransformation start doing the magic.
This class basically adds a method of signature
within MainTransformation.
Now within public static void main(String args, String args2), we declare a variable of type list containing 1, 2, 3 and this is being done by
Next we have added an if block , now if the block evaluates true then, we will declare another list and print Hello along with the list
and this is being done by:
block{
expression{
declaration {
variable "list2"
token "="
list {
constant "Ok1"
constant "Ok2"
constant "Ok333333"
}
}
}
expression { // NOTE: if block and else block are order dependent and same type
methodCall {
variable "this"
constant "println"
argumentList {
constant "Hello"
}
}
}
expression{
methodCall {
variable "this"
constant "println"
argumentList {
variable "list2"
}
}
}
}
and if the If block evaluated to false, then World will get printed.
The code can be tested as:
After we annotate the MainExample with @Main, and compile it, the bytecode of MainExample will contain all the intended changes.
Now the thing that I want to put stress here on is the way the Groovy transformation class implemented.
Here we have used new AstBuilder's buildFromSpec() method which takes a closure as an argument.
Now the closure call is being delgated to AstSpecificationCompiler class which contains all the necessary methods like method(), parameters(), block(), expression() etc..
Because:
public List buildFromSpec(@DelegatesTo(AstSpecificationCompiler.class) Closure specification)
{
........
}
So when the buildFromSpec() is invoked with the closure as the argument all the method calls are being delegated to AstSpecificationCompiler and each of the methods along with the parameters (if any) also contains a closure as an argument, which is again the method call to another method within AstSpecificationCompiler, i.e. Nested closures!!!!!! Intersting isn't it.
The way we have coded, Groovy Transformation class, is much like writing Business rules and less like coding, that's the beauty of Groovy DSL.
This feature of Groovy is really engaging, I must say, and with buildFromCode() Or buildFromString() we won't be getting this flavour.
In my next blog I will try to show to add more components within a class using Groovy ASTBuilder. Till then keep sharing and keep coding.....
