processor

Documentation

Overview

Package processor defines: 1. A TransactionHandler interface to be used to create new transaction families.

2. A high-level, general purpose, multi-threaded TransactionProcessor that any number of handlers can be added to.

3. A Context class used to abstract getting and setting addresses in global validator state.

Index

• Constants
• type Attribute
• type AuthorizationException
  • func (err *AuthorizationException) Error() string
• type Context
  • func NewContext(connection messaging.Connection, contextId string) *Context
  • func (self *Context) AddEvent(event_type string, attributes []Attribute, event_data []byte) error
  • func (self *Context) AddReceiptData(data []byte) error
  • func (self *Context) DeleteState(addresses []string) ([]string, error)
  • func (self *Context) GetState(addresses []string) (map[string][]byte, error)
  • func (self *Context) SetState(pairs map[string][]byte) ([]string, error)
• type InternalError
  • func (err *InternalError) Error() string
• type InvalidTransactionError
  • func (err *InvalidTransactionError) Error() string
• type TransactionHandler
• type TransactionProcessor
  • func NewTransactionProcessor(uri string) *TransactionProcessor
  • func (self *TransactionProcessor) AddHandler(handler TransactionHandler)
  • func (self *TransactionProcessor) SetMaxQueueSize(n uint)
  • func (self *TransactionProcessor) SetThreadCount(n uint)
  • func (self *TransactionProcessor) Shutdown()
  • func (self *TransactionProcessor) ShutdownOnSignal(siglist ...os.Signal)
  • func (self *TransactionProcessor) Start() error

Constants

const DEFAULT_MAX_WORK_QUEUE_SIZE = 100

Types

type Attribute

type Attribute struct {
	Key   string
	Value string
}

type AuthorizationException

type AuthorizationException struct {
	Msg          string
	ExtendedData []byte
}

func (*AuthorizationException) Error

func (err *AuthorizationException) Error() string

Raised when a authorization error occurs.

type Context

type Context struct {
	// contains filtered or unexported fields
}

Context provides an abstract interface for getting and setting validator state. All validator interactions by a handler should be through a Context instance. Currently, the Context class is NOT thread-safe and Context classes may not share the same messaging.Connection object.

func NewContext

func NewContext(connection messaging.Connection, contextId string) *Context

Construct a new context object given an initialized Stream and Context ID.

func (*Context) AddEvent

func (self *Context) AddEvent(event_type string, attributes []Attribute, event_data []byte) error

Add a new event to the execution result for this transaction.

func (*Context) AddReceiptData

func (self *Context) AddReceiptData(data []byte) error

Add a blob to the execution result for this transaction.

func (*Context) DeleteState

func (self *Context) DeleteState(addresses []string) ([]string, error)

DeleteState requests that each address in the validator state be deleted from state. A slice of addresses deleted is returned or an error if there was a problem deleting the addresses. For example:

responses, err := context.DeleteState(dataMap)
if err != nil {
    fmt.Println("Error deleting addresses!")
}
delete, ok := results[address]
if !ok {
    fmt.Prinln("Address was not deleted!")
}

func (*Context) GetState

func (self *Context) GetState(addresses []string) (map[string][]byte, error)

GetState queries the validator state for data at each of the addresses in the given slice. A string->[]byte map is returned. If an address is not set, it will not exist in the map.

results, err := context.Get(addresses)
if err != nil {
    fmt.Println("Error getting data!")
}
data, ok := results[address]
if !ok {
    fmt.Prinln("No data stored at address!")
}

func (*Context) SetState

func (self *Context) SetState(pairs map[string][]byte) ([]string, error)

SetState requests that each address in the validator state be set to the given value. A slice of addresses set is returned or an error if there was a problem setting the addresses. For example:

responses, err := context.Set(dataMap)
if err != nil {
    fmt.Println("Error setting addresses!")
}
set, ok := results[address]
if !ok {
    fmt.Prinln("Address was not set!")
}

type InternalError

type InternalError struct {
	Msg          string
	ExtendedData []byte
}

func (*InternalError) Error

func (err *InternalError) Error() string

Raised when an internal error occurs during transaction processing.

type InvalidTransactionError

type InvalidTransactionError struct {
	Msg          string
	ExtendedData []byte
}

func (*InvalidTransactionError) Error

func (err *InvalidTransactionError) Error() string

Raised for an Invalid Transaction.

type TransactionHandler

type TransactionHandler interface {
	// FamilyName should return the name of the transaction family that this
	// handler can process, e.g. "intkey"
	FamilyName() string

	// FamilyVersions should return the versions of the transaction
	// family that this handler can process. Eg., ["1.0", "1.5"]
	FamilyVersions() []string

	// Namespaces should return a slice containing all the handler's
	// namespaces, e.g. []string{"abcdef"}
	Namespaces() []string

	// Apply is the single method where all the business logic for a
	// transaction family is defined. The method will be called by the
	// transaction processor upon receiving a TpProcessRequest that the handler
	// understands and will pass in the TpProcessRequest and an initialized
	// instance of the Context type.
	Apply(*processor_pb2.TpProcessRequest, *Context) error
}

TransactionHandler is the interface that defines the business logic for a new transaction family. This is the only interface that needs to be implemented to create a new transaction family.

To create a transaction processor that uses a new transaction handler:

validatorEndpoint := "tcp://localhost:4004"
myHandler := NewMyHandler()
processor := NewTransactionProcessor(validatorEndpoint)
processor.AddHandler(myHandler)
processor.Start()

The FamilyName(), FamilyVersions(), and Namespaces() methods are used by the processor to route processing requests to the handler.

type TransactionProcessor

type TransactionProcessor struct {
	// contains filtered or unexported fields
}

TransactionProcessor is a generic class for communicating with a validator and routing transaction processing requests to a registered handler. It uses ZMQ and channels to handle requests concurrently.

func NewTransactionProcessor

func NewTransactionProcessor(uri string) *TransactionProcessor

NewTransactionProcessor initializes a new Transaction Process and points it at the given URI. If it fails to initialize, it will panic.

func (*TransactionProcessor) AddHandler

func (self *TransactionProcessor) AddHandler(handler TransactionHandler)

AddHandler adds the given handler to the TransactionProcessor so it can receive transaction processing requests. All handlers must be added prior to starting the processor.

func (*TransactionProcessor) SetMaxQueueSize

func (self *TransactionProcessor) SetMaxQueueSize(n uint)

func (*TransactionProcessor) SetThreadCount

func (self *TransactionProcessor) SetThreadCount(n uint)

Set the number of worker threads to be created for handling requests. Must be set before calling Start()

func (*TransactionProcessor) Shutdown

func (self *TransactionProcessor) Shutdown()

Shutdown sends a message to the processor telling it to deregister.

func (*TransactionProcessor) ShutdownOnSignal

func (self *TransactionProcessor) ShutdownOnSignal(siglist ...os.Signal)

ShutdownOnSignal sets up signal handling to shutdown the processor when one of the signals passed is received.

func (*TransactionProcessor) Start

func (self *TransactionProcessor) Start() error

Start connects the TransactionProcessor to a validator and starts listening for requests and routing them to an appropriate handler.