Scala Future with play framework

Async controller
----------------------------

package beard.controllers

import play.api.mvc._
import play.api.libs.json._
import play.api.libs.functional.syntax._
import play.libs.Json._
import reactivemongo.bson._
import service.BeardService
import play.api.libs.concurrent.Execution.Implicits.defaultContext

import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.concurrent.Future
import scala.util.{Failure, Success, Random}

class AsyncBeardController extends Controller {

 def sayAsyncBeard = Action.async { request =>
    val futureResult = Future {
      intensiveComputation()
    }
    futureResult.map(result =>
      Ok(result)
    )
  }


def intensiveComputation(): JsObject = {
      Thread.sleep(Random.nextInt(5000))
        Json.obj("value" -> "beard")
  }
}

Test
-------

$ curl -XGET http://localhost:9000/sayAsyncBeard
{"value":"beard"}

You will get the response after ~5seconds.

CPU cores in OSX

Seems OSX sucks in command line support as Linux does. However, found few commands to get hw information in OSX.

sysctl -n hw.ncpu
8

sysctl -n hw.physicalcpu
4

sysctl -n hw.logicalcpu
8


Ref : https://coolaj86.com/articles/get-a-count-of-cpu-cores-on-linux-and-os-x/

Kafka Events Streaming - Spark Consumer Example

* Apache Kafka supports a wide range of use cases as a general-purpose messaging system for scenarios where high throughput, reliable delivery, and horizontal scalability are important.
Use cases include:
* Stream Processing
* Website Activity Tracking
* Metrics Collection and Monitoring
* Log Aggs

* Apache Storm and Apache Spark both work very well in combination with Kafka.

Here's an example how Kafka can be used with apache spark for consuming events stream,

DEVELOPMENT
STEP 1 : ADD lib dependency to decoupled-invocation/build.sbt 

name := "decoupled-invocation"                                                                                                                        

version := "1.0"                                                                                    

scalaVersion := "2.10.4"                                                                            

val sparkVersion = "1.2.0"                                                                          

libraryDependencies ++= Seq(                                                                        
  "org.apache.spark" %% "spark-streaming" % sparkVersion % "provided",                              
  "org.apache.spark" %% "spark-streaming-kafka" % sparkVersion                                      
)   

STEP 2 : Create kafka stream in scala file decoupled-invocationsrc/main/scala/EventsConsumerApp.scala

object EventsConsumerApp {                                                                          

  def persist() = {                                                                                 

  }                                                                                                 

  def main(args : Array[String]): Unit = {                                                          
   val sparkConf = new SparkConf().setMaster("spark://prayagupd:7077").setAppName("EventsConsumerApp")
   val sparkStreamingContext = new StreamingContext(sparkConf, Seconds(10))                         

   val kafkaConf = Map("metadata.broker.list" -> "localhost:9092",                                  
                       "zookeeper.connect" -> "localhost:2181",                                     
                       "group.id" -> "events-topic-consumer-group",                                 
                       "zookeeper.connection.timeout.ms" -> "1000")                                 

    //http://spark.apache.org/docs/latest/streaming-programming-guide.html#discretized-streams-dstreams
   val kafkaDiscretizedStream = KafkaUtils.createStream[Array[Byte], String, DefaultDecoder, StringDecoder](sparkStreamingContext,
                                             kafkaConf,                                             
                                             Map("events-topic" -> 1), StorageLevel.MEMORY_ONLY_SER)
   //persist()                                                                                      
   kafkaDiscretizedStream.print()                                                                   
   sparkStreamingContext.start()                                                                    
   sparkStreamingContext.awaitTermination()                                                         
 }                                                                                                  
}   

STEP 3 : build application

sbt assembly ## doing sbt package wont find kafka jar while submitting job to spark







DEPLOYMENT
STEP 4 : start kafka broker with default config

cd /usr/local/kafka
bin/zookeeper-server-start.sh config/zookeeper.properties

#terminal 2
bin/kafka-server-start.sh config/server.properties

STEP 5 : Create kafka topic events-topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic events-topic
Created topic "events-topic".

verify topic is created

bin/kafka-topics.sh --list --zookeeper localhost:2181
events-topic

STEP 6 : Produce stream using kafka to topic events-topic

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic events-topic
{'message' : 'user is logging in'}
{'message' : 'User logged in'}

STEP 7 : submit spark job that will consume events produced by above kafka producer

/usr/local/spark-1.2.0/bin/spark-submit --class EventsConsumerApp --master spark://prayagupd:7077 target/scala-2.10/decoupled-invocation-assembly-1.0.jar

(null,{'message' : 'user is logging in'})
(null,{'message' : 'User logged in'})

15/09/20 17:04:12 INFO scheduler.JobScheduler: Finished job streaming job 1442786650000 ms.0 from job set of time 1442786650000 ms
15/09/20 17:04:12 INFO scheduler.JobScheduler: Total delay: 2.294 s for time 1442786650000 ms (execution: 2.289 s)
15/09/20 17:04:12 INFO rdd.BlockRDD: Removing RDD 5 from persistence list
15/09/20 17:04:12 INFO storage.BlockManager: Removing RDD 5
15/09/20 17:04:12 INFO kafka.KafkaInputDStream: Removing blocks of RDD BlockRDD[5] at createStream at EventsConsumerApp.scala:39 of time 1442786650000 ms
15/09/20 17:04:12 INFO scheduler.ReceivedBlockTracker: Deleting batches ArrayBuffer(1442786630000 ms)
15/09/20 17:04:12 INFO scheduler.ReceivedBlockTracker: Deleting batches ArrayBuffer()
15/09/20 17:04:20 INFO scheduler.JobScheduler: Added jobs for time 1442786660000 ms

As seen in console events produced using kafka-producer can be consumed in apache spark + kafka application, and can be persited to NoSQL databases or wherever thereafter.

Source code
scalability-patterns/decoupled-invocation

Resources
Spark Streaming + Kafka Integration Guide

git branching model

Git Branching most of the time seemed like a maze to me. Specially while sending a Pull Request for sprint development.

When there are three features in a sprint,

1) One idea could be  - maintain three local branches for each feature and send three parallel PRs separately.
After feature gets merged, delete local and remote feature branches.

2) The other could be - maintain three local branches for each feature, but when finished merge them to one single branch and send a PR one by one or ?.

I think the second one restricts from having number of unwanted branches to be deleted later, which looks OK to me while working in a team. 

Here's how I implement this approach; assuming sprint/develop is an upto-date branch created with HEAD of master,

STEP 1 - create feature branch

git checkout -b module-1/feature-1 sprint/develop

STEP 2 - after feature completion

Merge it to single sprint/develop branch

git checkout sprint/develop

git merge --no-ff module-1/feature-1 # create a new commit object.

## delete the feature branch

git branch -d module-1/feature-1

STEP 3 - rebase sprint/develop branch with master

git rebase master ## if any conflicts occur, resolve them manually, git add conflicted files, and

                            ## git rebase --continue

STEP 4 - push develop changes

git push origin sprint/develop  ## or git push --force origin sprint/develop

STEP 5 - Send PR and get reviewed
Send PR for merge on master. what?? few implementation not aligned with architecture of codebase?? FIX them on same branch and push.

A nerd's java practices, might not be universal though

Here's what a java nerd thinks on practices while coding. 
Purpose could be to 
* make beautiful code,
* improve perf
* improve readability etc

1. class members vs passing args between internal methods

Prefer class members or passing arguments between internal methods?

Do methods in class instances take a place in memory?

2. Temp variable in foreach iteration of a large loop

Temporary variable used for each iteration of a large loop, strings are immutable so what should I use?

3. Immutability habit (an affair with final) 
Immutability are to simplify a program, here's what the veterans have to say;

"Classes should be immutable unless there's a very good reason to make them mutable....If a class cannot be made immutable, limit its mutability as much as possible." - Joshua Bloch

"Immutability Fosters Concurrent Programming." - clojure

If immutable objects are good, why do people keep creating mutable objects?

Excessive use “final” keyword in Java

Wikileaks To Leak 5000 Open Source Java Projects With All That Private/Final Bullshit Removed

Side look on DSL

I’m a huge fan of DSL these days and try to implement wherever possible. I first heard of DSL through Hibernate query DSL, then groovy lang, Proto Buffers and so on.

Martin Fowler would define DSL as A a computer lang that's targeted to a particular kind of problem, rather than a general purpose lang that's aimed at any kind of software problem.

I found some libraries like hibernate dsl making their own embedded DSL, which we have to follow as it is. But there’s always a royal road for us to create our own DSL to solve our domain problems.

So, Internal DSLs aka embedded DSL aka Fluent Interface ( FI ) aka semantic facades are particular ways of using a host language (say Java) to give the host language the feel of a particular language.

eg. in Elasticsearch Java API,

           AggregationBuilder aggregationBuilder  =

                   (DateRangeBuilder)AggregationBuilders.dateRange("dateRangeAggs")

                                                     .field("transactionDate")

                                                     .addRange("Oct-1989", "1989-10-28", "1990-10-28")

                                                     .addRange("Nov-1990", "1990-10-28", "1991-10-28")

                                                     .subAggregation(AggregationBuilders.sum("transactionAmtSum")

                                                                                               .field("transactionAmt"))

                                                     .subAggregation(AggregationBuilders.avg("balanceAvg")

                                                                                                .field("balanceAmt"));

Internal DSLs are about readability.

Other internal DSL examples

Squill

Scala - Time DSL

While External DSLs have their own custom syntax and we write a full parser to process them.

eg.  1)

/**

 * cassandra config

 */

servers = {

   clusterName("gccount cluster")

   server(){

           name("Node1")

           hostname("localhost")

           port("9160")

           keyspace("gccount")

}

But we need a parser for above DSL, in groovy there’s groovy.util.BuilderSupport for creating arbitrary nested trees of objects. In Java we can write parsers as here.

eg. 2) clojureQl DSL
(select (table :Customer) (where (= :customerId 28)))

References

Creating DSLs in Java, Part 3: Internal and external DSLs

Harnessing Domain-Specific Languages (DSLs)

Scalding: Twitter's Scala DSL for Hadoop/Cascading

Hacking on MEAN stack

The Objective of this post is to get started with MEAN stack, where

M => mongodb
E  => expressjs
A  => angularjs
N  => nodejs

STEP 1 Install M(ongodb)
Mongodb is the a document database serving storage for MEAN stack.
To install it, please follow installation steps at Hacking on grails and mongodb

STEP 2 install N(odejs)
Node.js is a platform built on Chrome's JS runtime for easily building fast, scalable network apps.
To install it, please follow Hacking on node.js and geddy

STEP 3 Install E(xpressjs)
Express is a minimal and flexible node.js framework (for web application).
Assuming npm is installed, execute following command to install express.
$ npm install -g express

Seems it needs npm install express-generator -g to get express working in lates versions.

3.1 create express web app

$ express onlywallet

   create : onlywallet
   create : onlywallet/package.json
   create : onlywallet/app.js
   create : onlywallet/public
   create : onlywallet/public/images
   create : onlywallet/routes
   create : onlywallet/routes/index.js
   create : onlywallet/routes/user.js
   create : onlywallet/public/stylesheets
   create : onlywallet/public/stylesheets/style.css
   create : onlywallet/views
   create : onlywallet/views/layout.jade
   create : onlywallet/views/index.jade
   create : onlywallet/public/javascripts

   install dependencies:
     $ cd onlywallet && npm install

   run the app:
     $ node app

3.2 add mongodb driver and mongoosejs

"dependencies": {

  "express": "3.0.3",

  "jade": "*",

  "mongodb": ">= 0.9.6-7",

  "mongoose" : ">= 3.6"

}

3.3 configure app.js to connect to mongodb

var Mongoose = require('mongoose');

var db = Mongoose.createConnection('localhost', 'onlywallet');

STEP 4 install A(ngularjs) using Bower
AngularJS is for writing client-side web apps as if you had a smarter browser.

$ npm install bower –g
bower install angular#1.0.6

Just noticed http://mean.io, where MEAN comes as a single bundle within npm.

References
http://expressjs.com/guide.html

http://dandean.com/nodejs-npm-express-osx/

http://thecodebarbarian.wordpress.com/2013/07/22/introduction-to-the-mean-stack-part-one-setting-up-your-tools/