Forem: Saurabh Kishore Tiwari

Observer Pattern in Scala

Saurabh Kishore Tiwari — Sun, 30 Apr 2023 14:48:15 +0000

Observer pattern has two main components — Subject and Observer. All observers subscribe to the subject, and in case of any kind of change, the subject is responsible for propagating that information to all observers. As you may have guessed, it sounds like a one-to-many dependency between objects. Let’s have a look at the UML diagram to get a better understanding.

Let’s breakdown the structure now

Subject

Maintains a collection of all the observers.
Provides an interface to attach and detach the observers.

Observer

Defines an updating interface for objects that should be notified of changes in the subject.

ConcreteSubject

Stores the state of interest to ConcreteObserver.
Sends a notification to its observers when it’s state changes.

ConcreteObserver

Maintains a reference to ConcreteSubject.
Might store state that should stay consistent with the subject’s state.
Implements the Observer updating interface to keep its state consistent with the subject.

Implementation

Consider a stock company that tells its users the real-time value of all the stocks. The subject will be the stock company, and all the users who subscribe to it will be its observers.

CODE

User.scala

sealed trait Observer {
  def publishUpdate(stocks: mutable.Map[Int, Stock]): Unit
}

case class User(name: String, uuid: String) extends Observer:
  def publishUpdate(stocks: mutable.Map[Int, Stock]): Unit =
    println(name + " " + stocks.values.map(x => x.name + " " + x.price))

StockCompany.scala

case class Stock(id: Int, name: String, price: Double)

sealed trait Subject {
  def registerNewUser(user: User): Unit
  def notifyUser(): Unit
  def deleteUser(user: User): Unit
  def registerStocks(stock: Stock): Unit
  def updateStockPrice(stock: Stock): Unit
}

object StockCompany extends Subject:
  private val stocks = mutable.TreeMap[Int, Stock]()
  private val users = mutable.TreeMap[String, User]()

  def registerNewUser(user: User): Unit =  // Attach
    users.put(user.uuid, user)
    user.publishUpdate(stocks)             // As soonas a user registers they get the live prices

  def notifyUser(): Unit =                 // Notify
    users.foreach(_._2.publishUpdate(stocks))

  def deleteUser(user: User): Unit =       // Detach
    users.remove(user.uuid)

  def registerStocks(stock: Stock): Unit =
    stocks.put(stock.id, stock)
    notifyUser()

  def updateStockPrice(stock: Stock): Unit =
    stocks.put(stock.id, stock)
    notifyUser()

MainRunner.scala

object MainRunner:
  def main(args: Array[String]): Unit =
    val user1 = User("user1", "ADBPR4561E")
    val user2 = User("user2", "BFTPD3461S")

    val stock1 = Stock(1, "stock1", 23.42)
    val stock2 = Stock(2, "stock2", 34.53)
    val stock3 = Stock(3, "stock3", 45.64)

    StockCompany.registerStocks(stock1)
    StockCompany.registerStocks(stock2)

    StockCompany.registerNewUser(user1)
    Thread.sleep(1000)
    StockCompany.registerNewUser(user2)
    Thread.sleep(1000)
    StockCompany.registerStocks(stock3)
    Thread.sleep(1000)
    StockCompany.updateStockPrice(Stock(3, "stock3", 123.45))

Known Uses and Related Patterns

This pattern can be used in any scenario, when the user has to be notified in case of any event/change occurs, like:

Mobile app for weather update
Stock Prices getting updated live on WebPages
Ticket confirmation
Notification from any application
Following a user on Instagram 😂
“Notify me” when available on e-commerce sites
Event listener on a button

Now, if you give it a thought, the StockCompany has to take care of a lot of things. If we can somehow segregate the task and add another layer of abstraction (say a PriceManager) which will take care of notifying all the users, the stock company will just have to produce the prices to Price Manager and users will subscribe to Price manager while the Price manager orchestrates everything. This is kind of like a Publisher-Subscriber model(can be called a mediator model too based on the implementation)

Comma OK in GO

Saurabh Kishore Tiwari — Mon, 19 Dec 2022 17:36:19 +0000

Have you ever come across a statement like this?

func main(){
  i:= "hello world"
  n, ok := i.(int)
   if ok {
    fmt.Println("n :", n)
   } else {
    fmt.Println("not an int")
   }
}

Well, there’s no Try Catch block in Go. Coming from Java/Scala background, it felt weird at first but then GO grows on you, and then voila…you’re handling errors like a pro.

Comma ok is one such way of error handling. It’s used in 4 different scenarios:

Test for error on a function return
Testing if a key-value item exists in a Go map
Testing if an interface variable is of a certain type
Testing if a channel is closed

Let’s have a look at each one with an example

1. Test for error on a function return

func main() {
 if ok, err := isEven(3); ok {
  fmt.Println("it's an even number")
 }else{
  fmt.Println(err)
 }
}

func isEven(n int)(bool, error){
 if n & 1 == 1{
  return false, fmt.Errorf("it's an odd number")
 }
 return true, nil
}

The function isEven() takes an integer as input and checks if it’s even or not. The return type of isEven() is boolean and error. We store the boolean value in ok and use it in conditional statements.

2. Testing if a key-value item exists in a Go map

func main(){
   var stocks map[string]float64
   sym := "TTWO"
   price := stocks[sym]
   fmt.Printf("1. %s -> $%.2f\n", sym, price)

   if price, ok := stocks[sym]; ok {    // using comma ok to check if key exists
    fmt.Printf("2. %s -> $%.2f\n", sym, price)
   } else {
    fmt.Printf("2. %s not found\n", sym)
   }

   stocks = make(map[string]float64)   // It's importatnt to initialize the map before adding values. Else it's a panic situation
   stocks[sym] = 123.4
   stocks["APPL"] = 345.6

   if price, ok := stocks[sym]; ok {
    fmt.Printf("3. %s -> $%.2f\n", sym, price)
   } else {
    fmt.Printf("3. %s not found\n", sym)
   }
}

OUTPUT:
1. TTWO -> $0.00
2. TTWO not found
3. TTWO -> $123.40

Let’s understand what’s happening in the code.

At first, we declare a map stocks with (key, value) type as (string, float). We try to get the value for key=TTWO, as there’s no key that matches TTWO, the default value of float, i.e. 0.00 is returned.

Note that there’s no error that was returned. We actually don’t know if 0.00 is the assigned value or the default value.

This is where comma ok comes to the rescue. GO compiler is smart enough to understand what to return when the comma ok is added so in the line price, ok := stocks[sym], ok is assigned false, helping us with the dilemma.

3. Testing if an interface variable is of a certain type

func main(){
  i := "hello world"
  if n, ok := i.(int); ok{
    fmt.Println("It's an int")
  }else{
    fmt.Println("Not an int")
  }
}

This is just a slightly modified version of the code you saw above, to match the pattern we are following. That’s all 😉.
Notice how intelligent the GO-compiler is? It knows in what context the comma ok is being used.

4. Testing if a channel is closed

GO channel is a medium through which a goroutine communicates with another goroutine and this communication is lock-free

func main() {
 ch := make(chan string)    //initializing a channel
 go func() {
  for i := 1; i <= 3; i++ {
   msg := fmt.Sprintf("Current num is #%v", i)
   ch <- msg
  }
  close(ch) //closing the channel lets range know exactly ho much data is there
 }()

 for msg := range ch {
  fmt.Println("received message : ", msg)
 }

 msg := <-ch
 fmt.Printf("Message from a closed channel : %#v\n", msg)

 msg, ok := <-ch  // recieving value through a channel with comma ok
 fmt.Printf("Message : %#v, (was it closed? -> %#v)\n", msg, !ok)
}

OUTPUT : 
received message :  Current num is #1
received message :  Current num is #2
received message :  Current num is #3
Message from a closed channel : ""
Message : "", (was it closed? -> true)

We create a go routine, which pushes 3 integers into the channel ch and closes it before exiting the routine. The for loop consumes the data from channel ch. Then we try to assign the next message in the channel to msg. Since it’s empty we get the output as an empty string.

Channels don’t run into the panic situation if there's nothing to consume. They simply return a default value. But you cannot add any value into a closed channel. It gives rise to a panic situation.

We don’t really know if the value being consumed is the actual value or the default value. This is where comma ok comes to the rescue.

The Legends of GoRoutines — Parallelism and Concurrency

Saurabh Kishore Tiwari — Mon, 19 Dec 2022 11:58:17 +0000

Well, the clock speed of consumer CPUs has been constant to a few GHz for some years now. The obvious solution was to start gluing multiple CPU cores on top of each other to provide more processing power. This allowed for more available compute cycles.

This is an excellent approach if you want to run a lot of non-intensive CPU programs. The CPU scheduler works like a charm and distributes individual tasks over available CPU cores. But let’s say, we have just one job and wish to distribute it over multiple cores. This is where the design of the program and algorithm comes into the picture.

GO was designed for multitasking. There was so much importance given to multitasking that to start a parallel process in GO, you just need to start your statement with go. Take a look at the example below



func main(){
  for i := 1; i <= 3; i++{
    go func(n int){                // Parallel task is being created
      fmt.Printf("i : %d\n", n)
    }(i)
  } 
  time.Sleep(5 * time.Millisecond) // A sleep timer of 5 sec
}

Output: i : 3 i : 1 i : 2
Three parallel tasks are created and the user has no control over which one will execute first. Each parallel task is called a goroutine

If you don’t add a sleep timer, you probably won’t get any output because the execution won’t wait for a goroutine to complete. But adding a sleep timer isn’t a good idea, because the execution of routines isn’t something that can be predicted. A better way is to use sync.Waitgroup

GoRoutines Behind the Curtain

All goroutines are first pushed into a GRQ(Global Run Queue). Each Logical CPU has its OS Thread with its LRQ(Local Run Queue). Go Scheduler takes a routine from GRQ and randomly places it in any LRQ. Which is then picked up by the respective core and executed. This ensures Concurrency and Parallelism.

If you are from the distributed systems world like me, doesn’t it feel like the Go Scheduler is like the YARN scheduler in HADOOP ???

Special Cases (which are NOT so Special)

System calls(like I/O operations) are special cases where the execution of the program is suspended till the required operation is completed. If the routine stays in the LRQ till the operation is completed, a lot many CPU cycles will be wasted so a pretty decent solution is applied. Take an example of a single LRQ.

Say goroutine G7 encounters a syscall. So, instead of waiting, it’s sent to a place where it can stay till the syscall is over. It's a Parking space for goroutines which are

Sending and Receiving on Channel
Network I/O
System Call
Timers
Mutexes

Once ready for execution, G7 is sent to the LRQ again.