<commit_before>// Copyright (c) 2013 <NAME>
// Released under the MIT License
// http://opensource.org/licenses/MIT

package turnpike

import (
	"code.google.com/p/go.net/websocket"
	"encoding/json"
	"fmt"
	"github.com/nu7hatch/gouuid"
	"io"
	"log"
	"net"
	"net/http"
	"sync"
	"time"
)

var (
	// The amount of messages to buffer before sending to client.
	serverBacklog = 20
)

const (
	clientConnTimeout = 6
	clientMaxFailures = 3
)

// Server represents a WAMP server that handles RPC and pub/sub.
type Server struct {
	// Client ID -> send channel
	clients map[string]chan string
	// Client ID -> prefix mapping
	prefixes map[string]prefixMap
	// Proc URI -> handler
	rpcHandlers map[string]RPCHandler
	subHandlers map[string]SubHandler
	pubHandlers map[string]PubHandler
	// Topic URI -> subscribed clients
	subscriptions       map[string]listenerMap
	subLock             *sync.Mutex
	sessionOpenCallback func(string)
	websocket.Server
}

// RPCHandler is an interface that handlers to RPC calls should implement.
// The first parameter is the call ID, the second is the proc URI. Last comes
// all optional arguments to the RPC call. The return can be of any type that
// can be marshaled to JSON, or a error (preferably RPCError but any error works.)
// NOTE: this may be broken in v2 if multiple-return is implemented
type RPCHandler func(clientID string, topicURI string, args ...interface{}) (interface{}, error)

// RPCError represents a call error and is the recommended way to return an
// error from a RPC handler.
type RPCError struct {
	URI         string
	Description string
	Details     interface{}
}

// Error returns an error description.
func (e RPCError) Error() string {
	return fmt.Sprintf("turnpike: RPC error with URI %s: %s", e.URI, e.Description)
}

// SubHandler is an interface that handlers for subscriptions should implement to
// control with subscriptions are valid. A subscription is allowed by returning
// true or denied by returning false.
type SubHandler func(clientID string, topicURI string) bool

// PubHandler is an interface that handlers for publishes should implement to
// get notified on a client publish with the possibility to modify the event.
// The event that will be published should be returned.
type PubHandler func(topicURI string, event interface{}) interface{}

// NewServer creates a new WAMP server.
func NewServer() *Server {
	s := &Server{
		clients:       make(map[string]chan string),
		prefixes:      make(map[string]prefixMap),
		rpcHandlers:   make(map[string]RPCHandler),
		subHandlers:   make(map[string]SubHandler),
		pubHandlers:   make(map[string]PubHandler),
		subscriptions: make(map[string]listenerMap),
		subLock:       new(sync.Mutex),
	}
	s.Server = websocket.Server{
		Handshake: checkWAMPHandshake,
		Handler:   websocket.Handler(s.HandleWebsocket),
	}
	return s
}

// SetSessionOpenCallback adds a callback function that is run when a new session begins.
// The callback function must accept a string argument that is the session ID.
func (t *Server) SetSessionOpenCallback(f func(string)) {
	t.sessionOpenCallback = f
}

// RegisterRPC adds a handler for the RPC named uri.
func (t *Server) RegisterRPC(uri string, f RPCHandler) {
	if f != nil {
		t.rpcHandlers[uri] = f
	}
}

// UnregisterRPC removes a handler for the RPC named uri.
func (t *Server) UnregisterRPC(uri string) {
	delete(t.rpcHandlers, uri)
}

// RegisterSubHandler adds a handler called when a client subscribes to URI.
// The subscription can be canceled in the handler by returning false, or
// approved by returning true.
func (t *Server) RegisterSubHandler(uri string, f SubHandler) {
	if f != nil {
		t.subHandlers[uri] = f
	}
}

// UnregisterSubHandler removes a subscription handler for the URI.
func (t *Server) UnregisterSubHandler(uri string) {
	delete(t.subHandlers, uri)
}

// RegisterPubHandler adds a handler called when a client publishes to URI.
// The event can be modified in the handler and the returned event is what is
// published to the other clients.
func (t *Server) RegisterPubHandler(uri string, f PubHandler) {
	if f != nil {
		t.pubHandlers[uri] = f
	}
}

// UnregisterPubHandler removes a publish handler for the URI.
func (t *Server) UnregisterPubHandler(uri string) {
	delete(t.pubHandlers, uri)
}

// SendEvent sends an event with topic directly (not via Client.Publish())
func (t *Server) SendEvent(topic string, event interface{}) {
	t.handlePublish(topic, publishMsg{
		TopicURI: topic,
		Event:    event,
	})
}

// HandleWebsocket implements the go.net/websocket.Handler interface.
func (t *Server) HandleWebsocket(conn *websocket.Conn) {
	defer conn.Close()

	if debug {
		log.Print("turnpike: received websocket connection")
	}

	tid, err := uuid.NewV4()
	if err != nil {
		if debug {
			log.Print("turnpike: could not create unique id, refusing client connection")
		}
		return
	}
	id := tid.String()<commit_msg>Add a function to get the server's connected clients<commit_after>// Copyright (c) 2013 <NAME>
// Released under the MIT License
// http://opensource.org/licenses/MIT

package turnpike

import (
	"code.google.com/p/go.net/websocket"
	"encoding/json"
	"fmt"
	"github.com/nu7hatch/gouuid"
	"io"
	"log"
	"net"
	"net/http"
	"sync"
	"time"
)

var (
	// The amount of messages to buffer before sending to client.
	serverBacklog = 20
)

const (
	clientConnTimeout = 6
	clientMaxFailures = 3
)

// Server represents a WAMP server that handles RPC and pub/sub.
type Server struct {
	// Client ID -> send channel
	clients map[string]chan string
	// Client ID -> prefix mapping
	prefixes map[string]prefixMap
	// Proc URI -> handler
	rpcHandlers map[string]RPCHandler
	subHandlers map[string]SubHandler
	pubHandlers map[string]PubHandler
	// Topic URI -> subscribed clients
	subscriptions       map[string]listenerMap
	subLock             *sync.Mutex
	sessionOpenCallback func(string)
	websocket.Server
}

// RPCHandler is an interface that handlers to RPC calls should implement.
// The first parameter is the call ID, the second is the proc URI. Last comes
// all optional arguments to the RPC call. The return can be of any type that
// can be marshaled to JSON, or a error (preferably RPCError but any error works.)
// NOTE: this may be broken in v2 if multiple-return is implemented
type RPCHandler func(clientID string, topicURI string, args ...interface{}) (interface{}, error)

// RPCError represents a call error and is the recommended way to return an
// error from a RPC handler.
type RPCError struct {
	URI         string
	Description string
	Details     interface{}
}

// Error returns an error description.
func (e RPCError) Error() string {
	return fmt.Sprintf("turnpike: RPC error with URI %s: %s", e.URI, e.Description)
}

// SubHandler is an interface that handlers for subscriptions should implement to
// control with subscriptions are valid. A subscription is allowed by returning
// true or denied by returning false.
type SubHandler func(clientID string, topicURI string) bool

// PubHandler is an interface that handlers for publishes should implement to
// get notified on a client publish with the possibility to modify the event.
// The event that will be published should be returned.
type PubHandler func(topicURI string, event interface{}) interface{}

// NewServer creates a new WAMP server.
func NewServer() *Server {
	s := &Server{
		clients:       make(map[string]chan string),
		prefixes:      make(map[string]prefixMap),
		rpcHandlers:   make(map[string]RPCHandler),
		subHandlers:   make(map[string]SubHandler),
		pubHandlers:   make(map[string]PubHandler),
		subscriptions: make(map[string]listenerMap),
		subLock:       new(sync.Mutex),
	}
	s.Server = websocket.Server{
		Handshake: checkWAMPHandshake,
		Handler:   websocket.Handler(s.HandleWebsocket),
	}
	return s
}

// SetSessionOpenCallback adds a callback function that is run when a new session begins.
// The callback function must accept a string argument that is the session ID.
func (t *Server) SetSessionOpenCallback(f func(string)) {
	t.sessionOpenCallback = f
}

// RegisterRPC adds a handler for the RPC named uri.
func (t *Server) RegisterRPC(uri string, f RPCHandler) {
	if f != nil {
		t.rpcHandlers[uri] = f
	}
}

// UnregisterRPC removes a handler for the RPC named uri.
func (t *Server) UnregisterRPC(uri string) {
	delete(t.rpcHandlers, uri)
}

// RegisterSubHandler adds a handler called when a client subscribes to URI.
// The subscription can be canceled in the handler by returning false, or
// approved by returning true.
func (t *Server) RegisterSubHandler(uri string, f SubHandler) {
	if f != nil {
		t.subHandlers[uri] = f
	}
}

// UnregisterSubHandler removes a subscription handler for the URI.
func (t *Server) UnregisterSubHandler(uri string) {
	delete(t.subHandlers, uri)
}

// RegisterPubHandler adds a handler called when a client publishes to URI.
// The event can be modified in the handler and the returned event is what is
// published to the other clients.
func (t *Server) RegisterPubHandler(uri string, f PubHandler) {
	if f != nil {
		t.pubHandlers[uri] = f
	}
}

// UnregisterPubHandler removes a publish handler for the URI.
func (t *Server) UnregisterPubHandler(uri string) {
	delete(t.pubHandlers, uri)
}

// SendEvent sends an event with topic directly (not via Client.Publish())
func (t *Server) SendEvent(topic string, event interface{}) {
	t.handlePublish(topic, publishMsg{
		TopicURI: topic,
		Event:    event,
	})
}

// ConnectedClients returns a slice of the ids of all connected clients
func (t *Server) ConnectedClients() []string {
	clientIDs := []string{}
	for id, _ := range t.clients {
		clientIDs = append(clientIDs, id)
	}
	return clientIDs
}

// HandleWebsocket implements the go.net/websocket.Handler interface.
func (t *Server) HandleWebsocket(conn *websocket.Conn) {
	defer conn.Close()

	if debug {
		log.Print("turnpike: received websocket connection")
	}

	tid, err := uuid.NewV4()
	if err != nil {
		if debug {
			log.Print("turnpike: could not create unique id, refusing client connection")
		}
		return
	}
	id := tid.String()
{'fraction_non_alphanumeric': 0.09443507588532883, 'fraction_numerical': 0.002182323182223986, 'mean_word_length': 3.9494354442808053, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3887110', 'n_tokens_mistral': 3131, 'n_tokens_neox': 3028, 'n_words': 1160}