main-adapter-singule-bus.go

package main

type (
  Contextual interface {
        GetContext() interface{}
        SetContext() interface{}
  }

  Eventful interface {
    Consume(reporting chan interface{}, msg interface{})
        Produce(reporting chan interface{}, upstream chan interface{})
  }

  Handler func(...interface{}) error

  Reactive interface {
    GetHandlers() map[string]Handler
        SetHandler(string, Handler)
  }

  Module interface {
        Contextual
        Eventful
        Reactive
  }

  Controllable interface {
        Init()
        Start()
        Stop()
  }

  Roundabout interface {
    Sink()
  }

  Engine interface {
        Contextual
        Roundabout
        Controllable
  }

  NetworkAdapter interface {
     Broadcast()
         Receive()
         Send()
  }

  ConsensusAdapter interface {
        Controllable
        InitGenesis()
  }

  PersistenceAdapter interface {
  }

  eventStore struct {
    events []interface{}
        upstream chrn interface{}
        control chan int
  }

  engine struct {
        eventStore
    modules map[string]Module
  }

  utility struct {
  }

  consensus struct {
  }

  p2p struct {
        peerList map[string]Peer
        // other stuff
  }

  persistence struct {
  }
)

/**
 * Engine
 */
func (e *engine) Init(c *Container) {
  e.Modules = map[string]Modules{
        "Utility": Container.Get('utility'),
        "Consensus": Container.Get('consensus'),
        "P2p": Container.Get('p2p'),
        "Persistence": Container.Get('persistence'),
  }
}

/**

 * A scenario of the lifecyle is then a succession of events.
 * To Sync for instance, the events happen in the following sequence:
 *   Consensus.Sync -> P2p.GetBlock -> P2p.ReceivedBlock -> Consensus.ApplyBlock -> Utility.BeginBlock -> Utility.DeliverTxs -> Utility.EndBlock
 */
func (e *engine) Sink() {
  for {
        select {
          case msg := <-e.upstream:

                switch msg.Topic {
                  case ControlSignalStop:
                        break;

                  case SyncGetBlocks:
                        e.Modules['P2p'].Consume(msg, action='GetBlock')

                  case SyncReceivedBlocks:
                        e.Modules['Consensus'].Consume(msg, action='ApplyBlock')

                  case SyncBlocksInit:
                        e.Modules['Utility'].Consume(msg, action='BeginBlock')

                  case SyncBlocksInProgress:
                        e.Modules['Utility'].Consume(msg, action='DeliverTxs')

                  case SyncBlocksFinish:
                        e.Modules['Utility'].Consume(msg, action='EndBlock')

          case ProposeBlock:
                        e.Modules['Consensus'].Consume(msg, action='StartProposalRound')

                  case PropagateBlock:
                        e.Modules['P2p'].Consume(msg, action='BroadcastBlock')

                  case PropagateTxs:
                        e.Modules['P2p'].Consume(msg, action='BroadcastTx')

                  case ReceivedPropagatedTxs:
                        e.Modules['Consensus'].Consume(msg, action='')
                  }
          }
        }
}

func (e *engine) Start() {
  for {
        select {
          case signal := <-e.control:
                switch signal {
                  case ControlSignalStart:
                        go e.Init()
                        go e.Sink()
                        go e.Modules['P2p'].Start()
                        go e.Modules['Consensus'].Start()

          case ControlSignalStop:
                        e.upstream <- MsgControlDomainStop
                }
        }
  }
}

func (e *engine) Stop() {
  e.upstream <- MsgControlDomainStop
}

/**
 * Consensus
 */
func (c *consensus) Consume(reporting chan interface{}, msg interface{}, action Action) {
  if h, exists := c.Handlers[action]; exists {
    h(c, msg, action)
  } else {
    reporting <- "Unknown topic"
  }
}

func (c *consensus) Produce(reporting chan interface{}, msg Message) {
  c.GetContext().upstream <- msg
}

func (c *consensus) InitGensis(...) {
  c.Produce(SyncReceivedBlocks)
}
/**
 * Utility
 */
func (u *utility) Consume(reporting chan interface{}, msg interface{}) {
  if h, exists := c.Handlers[action]; exists {
    h(c, msg, action) // BeginBlock
  } else {
    reporting <- "Unknown topic"
  }
}

func (u *utility) Produce(reporting chan interface{}) upstream chan interface{} {
}



/**
 * p2p
 */
func NewNetworkAdapter() *p2p {
  p2p := &p2p{}

  p2p.Handlers['Receive'] = p2p.Receive
  // or some closure
  p2p.Handlers['Broadcast'] = func() {
        // stuff
        p2p.Broadcast()
        // stuff
  }
  //or somet generator pattern
  p2p.Handlers['Send'] = Middleware.WrapWithLogging(p2p.Send)
}

func (pp *p2p) Consume(reporting chan interface{}, msg interface{}) {
}

func (pp *p2p) Produce(reporting chan interface{}) upstream chan interface{} {
}

func (pp *p2p) Broadcast() {

}

func (pp *p2p) Receive() {

}

func (pp *p2p) Send() {

}

func (pp *p2p) SetHandler(actionName string, actionHandler Handler) {
  pp.Handlers[actionName] = actionHandler
}

func (pp *p2p) GetHandlers() {
  return pp.Handlers
}

/**
 * Persistence
 */
func (ps *persistence) Consume(reporting chan interface{}, msg interface{}) {
}

func (ps *persistence) Produce(reporting chan interface{}) upstream chan interface{} {
}