funmow/event_distributor.go

package funmow

const (
	EventTypeEmit  = iota // everyone in the containing object hears it,
	EventTypeOEmit        // everyone in the containing object except the player hears it
	EventTypePEmit        // only the player hears it
	EventTypeWall         // broadcast message
	EventTypePage
	EventTypeSay
	EventTypePose
	EventTypeCommand //7
	EventTypeOutput  //8
	EventTypeQuit
	EventTypeTeardown
	EventTypeTeardownComplete
	EventTypeForce
	EventTypeLogin
	EventTypeSystem
)

type PlayerEvent struct {
	connectionID int
	src          DBRef
	dst          DBRef
	messageType  int
	message      string
}

type playerRegistration struct {
	execContext *ExecutionContext
	connInbound map[int]chan PlayerEvent
	execInbound chan PlayerEvent
}

type EventSubscribeRequest struct {
	connectionID int
	playerID     DBRef
	inbound      chan PlayerEvent
	outbound     chan chan PlayerEvent
}

type EventDistributor struct {
	subscribeChan chan EventSubscribeRequest
	db            *DB
	players       map[DBRef]*playerRegistration
	whoChan       chan chan DBRefList
}

func NewEventDistributor(db *DB) *EventDistributor {
	e := new(EventDistributor)
	e.db = db
	e.subscribeChan = make(chan EventSubscribeRequest)
	e.whoChan = make(chan chan DBRefList)
	return e
}

func (e *EventDistributor) Subscribe(req EventSubscribeRequest) chan PlayerEvent {
	e.subscribeChan <- req
	outbound := <-req.outbound
	return outbound
}

func (e *EventDistributor) OnlinePlayers() DBRefList {
	replyChan := make(chan DBRefList)
	e.whoChan <- replyChan
	onlinePlayers := <-replyChan
	return onlinePlayers
}

//[client runs quit command]
//connection -> "quit" -> exec
//exec -> EventTypeQuit -> connection
//[connection stops doing connectiony stuff]
//connection -> EventTypeTeardown -> exec
//[exec stops, kills off goroutines]
//exec -> EventTypeTeardownComplete -> Event Distributor
//Event Distributor cleans up

//[client dies]
//connection -> EventTypeTeardown -> exec
//[exec stops, kills off goroutines]
//exec -> EventTypeTeardownComplete -> Event Distributor
//Event Distributor cleans up

func (e *EventDistributor) Run() {

	//ipcChans := make(map[int]playerRegistration)

	e.players = make(map[DBRef]*playerRegistration)

	outbound := make(chan PlayerEvent)
	outboundBuffered := EventBuffer(outbound)

	forceContext := NewForceContext(e, e.db, outbound)
	forceInbound := forceContext.StartInboundChannel()

	for {

		select {
		case replyChan := <-e.whoChan:
			onlinePlayers := make(DBRefList, 0)
			for playerID, _ := range e.players {
				onlinePlayers = append(onlinePlayers, playerID)
			}
			replyChan <- onlinePlayers
		case sub := <-e.subscribeChan:

			if _, exists := e.players[sub.playerID]; !exists {
				c := NewExecutionContext(sub.playerID, e, e.db, outbound)
				e.players[sub.playerID] = &playerRegistration{
					execInbound: c.StartInboundChannel(),
					execContext: c,
					connInbound: make(map[int]chan PlayerEvent),
				}
				// mark player as online, and tell everyone!
				e.players[sub.playerID].execInbound <- PlayerEvent{messageType: EventTypeLogin}
			}
			reg := e.players[sub.playerID]
			reg.connInbound[sub.connectionID] = sub.inbound
			sub.outbound <- outbound
		case event := <-outboundBuffered: // message
			//fmt.Printf("received message src:%d dest:%d type: %d\n", event.src, event.dst, event.messageType)

			destPlayer, validDest := e.players[event.dst]

			switch event.messageType {
			case EventTypeForce:
				forceInbound <- event
			case EventTypeTeardownComplete:
				if validDest {
					delete(e.players, event.dst)
				}
			case EventTypeOutput: // from context to connection
				if validDest {
					e.sendToAllConnections(destPlayer.connInbound, event)
				}
				//destPlayer.connInbound <- event
			case EventTypeQuit: // from context to connection
				if validDest {
					e.players[event.dst].connInbound[event.connectionID] <- event
				}
			case EventTypeTeardown: // comes from client when connection is dropped
				if validDest {
					close(e.players[event.dst].connInbound[event.connectionID])
					delete(e.players[event.dst].connInbound, event.connectionID)
					if len(e.players[event.dst].connInbound) == 0 {
						close(e.players[event.dst].execInbound) // closing the inbound channel signals the exec to stop
					}
				}
			case EventTypeCommand: // from connection to context
				if validDest {
					destPlayer.execInbound <- event
				}
			case EventTypePEmit:
				if validDest {
					destPlayer.execInbound <- event
				}
			case EventTypePage:
				if validDest {
					destPlayer.execInbound <- event
				}
			case EventTypeEmit:
				e.broadcastEvent(event, false)
			case EventTypeOEmit:
				e.broadcastEvent(event, true) // fix
			case EventTypeSay:
				e.broadcastEvent(event, true)
			case EventTypePose:
				e.broadcastEvent(event, false)
			case EventTypeWall:
				e.broadcastEvent(event, false)
			case EventTypeSystem:
				e.broadcastEvent(event, false)
			}
		}
	}
}

func (e *EventDistributor) sendToAllConnections(connections map[int]chan PlayerEvent, event PlayerEvent) {

	for _, channel := range connections {
		channel <- event
	}

}

func (e *EventDistributor) broadcastEvent(event PlayerEvent, omitSrc bool) {
	for playerID, player := range e.players {
		if !omitSrc || playerID != event.src {
			player.execInbound <- event
		}
	}
}