lugu/qiloop

package bus

import (
    "errors"
    "log"
    gonet "net"
    "sync"

    "github.com/lugu/qiloop/bus/net"
)

// ErrServiceNotFound is returned with a message refers to an unknown
// service.
var ErrServiceNotFound = errors.New("Service not found")

// ErrObjectNotFound is returned with a message refers to an unknown
// object.
var ErrObjectNotFound = errors.New("Object not found")

// ErrActionNotFound is returned with a message refers to an unknown
// action.
var ErrActionNotFound = errors.New("Action not found")

// ErrNotAuthenticated is returned with a message tries to contact a
// service without prior authentication.
var ErrNotAuthenticated = errors.New("Not authenticated")

// ErrTerminate is returned with an object lifetime ends while
// clients subscribes to its signals.
var ErrTerminate = errors.New("Object terminated")

// Activation is sent during activation: it informs the object of the
// context in which the object is being used.
type Activation struct {
    ServiceID uint32
    ObjectID  uint32
    Session   Session
    Terminate func()
    Service   Service
}

// Receiver handles incomming messages.
type Receiver interface {
    Receive(m *net.Message, from Channel) error
}

// Actor interface used by Server to manipulate services.
type Actor interface {
    Receiver
    Activate(activation Activation) error
    OnTerminate()
}

// firewall ensures an endpoint talks only to autorized services.
// Especially, it ensure authentication is passed.
func firewall(m *net.Message, from Channel) error {
    if from.Authenticated() == false && m.Header.Service != 0 {
        return ErrNotAuthenticated
    }
    return nil
}

// server listen from incomming connections, set-up the end points and
// forward the EndPoint to the dispatcher.
type server struct {
    listen        net.Listener
    addrs         []string
    namespace     Namespace
    Router        *Router
    contexts      map[Channel]bool
    contextsMutex sync.Mutex
    closeChan     chan int
    waitChan      chan error
}

// NewServer creates a new server which respond to incomming
// connection requests.
func NewServer(listener net.Listener, auth Authenticator,
    namespace Namespace, service1 Actor) (Server, error) {

    service0 := ServiceAuthenticate(auth)

    s := &server{
        listen:        listener,
        namespace:     namespace,
        contexts:      make(map[Channel]bool),
        contextsMutex: sync.Mutex{},
        closeChan:     make(chan int, 1),
        waitChan:      make(chan error, 1),
    }
    s.Router = NewRouter(service0, namespace, s.Session())
    _, err := s.NewService("ServiceDirectory", service1)
    if err != nil {
        s.Terminate()
        return nil, err
    }

    go s.run()
    return s, nil
}

// StandAloneServer starts a new server
func StandAloneServer(listener net.Listener, auth Authenticator,
    namespace Namespace) (Server, error) {

    service0 := ServiceAuthenticate(auth)

    s := &server{
        listen:        listener,
        namespace:     namespace,
        contexts:      make(map[Channel]bool),
        contextsMutex: sync.Mutex{},
        closeChan:     make(chan int, 1),
        waitChan:      make(chan error, 1),
    }
    s.Router = NewRouter(service0, namespace, s.Session())
    go s.run()
    return s, nil
}

// Service represents a running service.
type Service interface {
    ServiceID() uint32
    Add(o Actor) (uint32, error)
    Remove(objectID uint32) error
    Terminate() error
}

// ServiceReceiver represents the service used by the Server.
type ServiceReceiver interface {
    Service
    Receiver
}

// NewService returns a new service. The service is activated as part
// of the creation.
// This brings the new service online. The steps involves are:
// 1. request the name to the namespace (service directory)
// 2. activate the service
// 3. add the service to the router dispatcher
// 4. advertize the service to the namespace (service directory)
func (s *server) NewService(name string, object Actor) (Service, error) {

    s.Router.RLock()
    session := s.Router.session
    s.Router.RUnlock()

    // 1. reserve the name
    serviceID, err := s.namespace.Reserve(name)
    if err != nil {
        return nil, err
    }

    // 2. activate the service
    activation := serviceActivation(s.Router, session, serviceID)
    service, err := NewService(object, activation)
    if err != nil {
        return nil, err
    }

    // 3. make it available
    err = s.Router.Add(serviceID, service)
    if err != nil {
        return nil, err
    }

    // 4. advertize it
    err = s.namespace.Enable(serviceID)
    if err != nil {
        s.Router.Remove(serviceID)
        return nil, err
    }
    return service, nil
}

func (s *server) handle(stream net.Stream, authenticated bool) {

    context := &channel{
        capability: DefaultCap(),
    }
    if authenticated {
        context.SetAuthenticated()
    }
    filter := func(hdr *net.Header) (matched bool, keep bool) {
        if hdr.Type == net.Reply || hdr.Type == net.Error ||
            hdr.Type == net.Event || hdr.Type == net.Cancelled {
            return false, true
        }
        return true, true
    }
    consumer := make(chan *net.Message, 10)
    go func() {
        for msg := range consumer {
            err := firewall(msg, context)
            if err != nil {
                log.Printf("missing authentication from %s: %#v",
                    context.EndPoint().String(), msg.Header)
                context.SendError(msg, err)
                stream.Close()
                return
            }
            err = s.Router.Receive(msg, context)
            if err != nil {
                log.Printf("error %v: %s", msg.Header, err)
            }
        }
    }()
    closer := func(err error) {
        s.contextsMutex.Lock()
        defer s.contextsMutex.Unlock()
        if _, ok := s.contexts[context]; ok {
            delete(s.contexts, context)
        }
    }
    finalize := func(e net.EndPoint) {
        context.endpoint = e
        e.MakeHandler(filter, consumer, closer)
        s.contextsMutex.Lock()
        s.contexts[context] = true
        s.contextsMutex.Unlock()
    }
    net.EndPointFinalizer(stream, finalize)
}

func (s *server) run() {
    for {
        stream, err := s.listen.Accept()
        if err != nil {
            select {
            case <-s.closeChan:
            default:
                s.listen.Close()
                s.stoppedWith(err)
            }
            break
        }
        s.handle(stream, false)
    }
}

func (s *server) stoppedWith(err error) {
    // 1. informs all services
    s.Router.Terminate()
    // 2. close all connections
    s.closeAll()
    // 3. inform server's user
    s.waitChan <- err
    close(s.waitChan)
}

// CloseAll close the connecction. Return the first error if any.
func (s *server) closeAll() error {
    var ret error
    s.contextsMutex.Lock()
    defer s.contextsMutex.Unlock()
    for context := range s.contexts {
        err := context.EndPoint().Close()
        if err != nil && ret == nil {
            ret = err
        }
    }
    return ret
}

// WaitTerminate blocks until the server has terminated.
func (s *server) WaitTerminate() chan error {
    return s.waitChan
}

// Terminate stops a server.
func (s *server) Terminate() error {
    close(s.closeChan)
    err := s.listen.Close()
    s.stoppedWith(err)
    return err
}

// Client returns a local client able to contact services without
// creating a new connection.
func (s *server) Client() Client {
    ctl, srv := gonet.Pipe()
    s.handle(net.ConnStream(srv), true)
    return NewClient(NewChannel(net.ConnEndPoint(ctl), DefaultCap()))
}

// Session returns a local session able to contact local services
// without creating a new connection to the server.
func (s *server) Session() Session {
    return s.namespace.Session(s)
}