Merge remote-tracking branch 'upstream/bolt' into bolt

This commit is contained in:
Daniel Oliveira 2015-04-24 18:21:19 -06:00
commit 707b0e3e13
10 changed files with 175 additions and 128 deletions

View file

@ -1,89 +0,0 @@
package cluster
import (
"sync"
"github.com/samalba/dockerclient"
)
// TODO (bradrydzewski) ability to cancel work.
// TODO (bradrydzewski) ability to remove a worker.
type Cluster struct {
sync.Mutex
clients map[dockerclient.Client]bool
clientc chan dockerclient.Client
}
func New() *Cluster {
return &Cluster{
clients: make(map[dockerclient.Client]bool),
clientc: make(chan dockerclient.Client, 999),
}
}
// Allocate allocates a client to the pool to
// be available to accept work.
func (c *Cluster) Allocate(cli dockerclient.Client) bool {
if c.IsAllocated(cli) {
return false
}
c.Lock()
c.clients[cli] = true
c.Unlock()
c.clientc <- cli
return true
}
// IsAllocated is a helper function that returns
// true if the client is currently allocated to
// the Pool.
func (c *Cluster) IsAllocated(cli dockerclient.Client) bool {
c.Lock()
defer c.Unlock()
_, ok := c.clients[cli]
return ok
}
// Deallocate removes the worker from the pool of
// available clients. If the client is currently
// reserved and performing work it will finish,
// but no longer be given new work.
func (c *Cluster) Deallocate(cli dockerclient.Client) {
c.Lock()
defer c.Unlock()
delete(c.clients, cli)
}
// List returns a list of all Workers currently
// allocated to the Pool.
func (c *Cluster) List() []dockerclient.Client {
c.Lock()
defer c.Unlock()
var clients []dockerclient.Client
for cli := range c.clients {
clients = append(clients, cli)
}
return clients
}
// Reserve reserves the next available worker to
// start doing work. Once work is complete, the
// worker should be released back to the pool.
func (p *Cluster) Reserve() <-chan dockerclient.Client {
return p.clientc
}
// Release releases the worker back to the pool
// of available workers.
func (c *Cluster) Release(cli dockerclient.Client) bool {
if !c.IsAllocated(cli) {
return false
}
c.clientc <- cli
return true
}

View file

@ -114,6 +114,12 @@ func main() {
queue.POST("/push/:owner/:name/:build/:task/logs", server.PushLogs) queue.POST("/push/:owner/:name/:build/:task/logs", server.PushLogs)
} }
events := api.Group("/stream")
{
events.GET("/user", server.GetEvents)
//events.GET("/logs/:owner/:name/:build/:number")
}
auth := r.Group("/authorize") auth := r.Group("/authorize")
{ {
auth.Use(server.SetHeaders()) auth.Use(server.SetHeaders())

View file

@ -3,9 +3,9 @@ package eventbus
import "github.com/drone/drone/common" import "github.com/drone/drone/common"
type Event struct { type Event struct {
Build *common.Build Build *common.Build `json:"build,omitempty"`
Repo *common.Repo Repo *common.Repo `json:"repo,omitempty"`
Task *common.Task Task *common.Task `json:"task,omitempty"`
} }
type Bus interface { type Bus interface {

View file

@ -37,6 +37,7 @@ func PushBuild(c *gin.Context) {
c.Fail(404, err) c.Fail(404, err)
return return
} }
build.Duration = in.Duration build.Duration = in.Duration
build.Started = in.Started build.Started = in.Started
build.Finished = in.Finished build.Finished = in.Finished
@ -47,16 +48,33 @@ func PushBuild(c *gin.Context) {
return return
} }
if repo.Last == nil || build.Number >= repo.Last.Number {
repo.Last = build
store.SetRepo(repo)
}
// <-- FIXME
// for some reason the Repo and Build fail to marshal to JSON.
// It has something to do with memory / pointers. So it goes away
// if I just refetch these items. Needs to be fixed in the future,
// but for now should be ok
repo, err = store.Repo(repo.FullName)
if err != nil {
c.Fail(500, err)
return
}
build, err = store.Build(repo.FullName, in.Number)
if err != nil {
c.Fail(404, err)
return
}
// END FIXME -->
bus.Send(&eventbus.Event{ bus.Send(&eventbus.Event{
Build: build, Build: build,
Repo: repo, Repo: repo,
}) })
if repo.Last != nil && repo.Last.Number > build.Number {
c.Writer.WriteHeader(200)
return
}
repo.Last = build
store.SetRepo(repo)
c.Writer.WriteHeader(200) c.Writer.WriteHeader(200)
} }
@ -83,6 +101,7 @@ func PushTask(c *gin.Context) {
bus.Send(&eventbus.Event{ bus.Send(&eventbus.Event{
Build: build, Build: build,
Repo: repo, Repo: repo,
Task: in,
}) })
c.Writer.WriteHeader(200) c.Writer.WriteHeader(200)
} }

View file

@ -31,6 +31,7 @@
<script src="/static/scripts/services/users.js"></script> <script src="/static/scripts/services/users.js"></script>
<script src="/static/scripts/services/logs.js"></script> <script src="/static/scripts/services/logs.js"></script>
<script src="/static/scripts/services/tokens.js"></script> <script src="/static/scripts/services/tokens.js"></script>
<script src="/static/scripts/services/feed.js"></script>
<script src="/static/scripts/filters/filter.js"></script> <script src="/static/scripts/filters/filter.js"></script>
<script src="/static/scripts/filters/gravatar.js"></script> <script src="/static/scripts/filters/gravatar.js"></script>

View file

@ -4,7 +4,7 @@
* BuildsCtrl responsible for rendering the repo's * BuildsCtrl responsible for rendering the repo's
* recent build history. * recent build history.
*/ */
function BuildsCtrl($scope, $routeParams, builds, repos, users) { function BuildsCtrl($scope, $routeParams, builds, repos, users, feed) {
var owner = $routeParams.owner; var owner = $routeParams.owner;
var name = $routeParams.name; var name = $routeParams.name;
@ -40,12 +40,31 @@
delete $scope.repo.subscription; delete $scope.repo.subscription;
}); });
} }
feed.subscribe(function(event) {
if (event.repo.full_name !== fullName) {
return; // ignore
}
// update repository
$scope.repo = event.repo;
$scope.apply();
if (event.build.number !== parseInt(number)) {
return; // ignore
}
// update the build status
$scope.build.state = event.build.state;
$scope.build.started = event.build.started;
$scope.build.finished = event.build.finished;
$scope.build.duration = event.build.duration;
$scope.$apply();
});
} }
/** /**
* BuildCtrl responsible for rendering a build. * BuildCtrl responsible for rendering a build.
*/ */
function BuildCtrl($scope, $routeParams, logs, builds, repos, users) { function BuildCtrl($scope, $routeParams, logs, builds, repos, users, feed) {
var step = parseInt($routeParams.step) || 1; var step = parseInt($routeParams.step) || 1;
var number = $routeParams.number; var number = $routeParams.number;
@ -99,6 +118,32 @@
}); });
}; };
feed.subscribe(function(event) {
if (event.repo.full_name !== fullName) {
return; // ignore
}
if (event.build.number !== parseInt(number)) {
return; // ignore
}
// update the build status
$scope.build.state = event.build.state;
$scope.build.started = event.build.started;
$scope.build.finished = event.build.finished;
$scope.build.duration = event.build.duration;
$scope.$apply();
if (!event.task || event.task.number !== step) {
return; // ignore
}
// update the task status
$scope.task.state = event.task.state;
$scope.task.started = event.task.started;
$scope.task.finished = event.task.finished;
$scope.task.duration = event.task.duration;
$scope.task.exit_code = event.task.exit_code;
$scope.$apply();
});
// var convert = new Filter({stream:true,newline:false}); // var convert = new Filter({stream:true,newline:false});
// var term = document.getElementById("term") // var term = document.getElementById("term")
// var stdout = document.getElementById("stdout").innerText.split("\n") // var stdout = document.getElementById("stdout").innerText.split("\n")

View file

@ -4,7 +4,7 @@
* ReposCtrl responsible for rendering the user's * ReposCtrl responsible for rendering the user's
* repository home screen. * repository home screen.
*/ */
function ReposCtrl($scope, $routeParams, repos, users) { function ReposCtrl($scope, $routeParams, repos, users, feed) {
// Gets the currently authenticated user // Gets the currently authenticated user
users.getCached().then(function(payload){ users.getCached().then(function(payload){
@ -18,6 +18,19 @@
}).catch(function(err){ }).catch(function(err){
$scope.error = err; $scope.error = err;
}); });
feed.subscribe(function(event) {
if (!$scope.repos) {
return;
}
for (var i=0;i<$scope.repos.length;i++) {
if ($scope.repos[i].full_name === event.repo.full_name) {
$scope.repos[i]=event.repo;
$scope.$apply();
break;
}
};
});
} }
/** /**

View file

@ -117,24 +117,23 @@
}); });
} }
// /**
// *
// */
// function RouteChange($rootScope, stdout, projs) {
// $rootScope.$on('$routeChangeStart', function (event, next) {
// projs.unsubscribe();
// stdout.unsubscribe();
// });
// //$rootScope.$on('$routeChangeSuccess', function (event, current, previous) { function RouteChange($rootScope, feed) {
// // document.title = current.$$route.title + ' · drone.io'; $rootScope.$on('$routeChangeStart', function (event, next) {
// //}); feed.unsubscribe();
// } });
$rootScope.$on('$routeChangeSuccess', function (event, current, previous) {
if (current.$$route.title) {
document.title = current.$$route.title + ' · drone';
}
});
}
angular angular
.module('drone') .module('drone')
.config(Authorize) .config(Authorize)
.config(Config); .config(Config)
// .run(RouteChange); .run(RouteChange);
})(); })();

View file

@ -0,0 +1,31 @@
'use strict';
(function () {
function FeedService($http, $window) {
var token = localStorage.getItem('access_token');
var proto = ($window.location.protocol == 'https:' ? 'wss' : 'ws');
var route = [proto, "://", $window.location.host, '/api/stream/user?access_token=', token].join('');
var wsCallback = undefined;
var ws = new WebSocket(route);
ws.onmessage = function(event) {
var data = angular.fromJson(event.data);
if (wsCallback != undefined) {
wsCallback(data);
}
};
this.subscribe = function(callback) {
wsCallback = callback;
};
this.unsubscribe = function() {
wsCallback = undefined;
};
}
angular
.module('drone')
.service('feed', FeedService);
})();

View file

@ -3,8 +3,10 @@ package server
import ( import (
"time" "time"
"github.com/drone/drone/common"
"github.com/drone/drone/eventbus" "github.com/drone/drone/eventbus"
log "github.com/Sirupsen/logrus"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
) )
@ -32,6 +34,16 @@ func GetEvents(c *gin.Context) {
user := ToUser(c) user := ToUser(c)
remote := ToRemote(c) remote := ToRemote(c)
// TODO (bradrydzewski) revisit this approach at some point.
//
// instead of constantly checking for remote permissions, we will
// cache them for the lifecycle of this websocket. The pro here is
// that we are making way less external calls (good). The con is that
// if a ton of developers conntect to websockets for long periods of
// time with heavy build traffic (not super likely, but possible) this
// caching strategy could take up a lot of memory.
perms_ := map[string]*common.Perm{}
// upgrade the websocket // upgrade the websocket
ws, err := upgrader.Upgrade(c.Writer, c.Request, nil) ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil { if err != nil {
@ -56,15 +68,24 @@ func GetEvents(c *gin.Context) {
if event == nil { if event == nil {
return // why would this ever happen? return // why would this ever happen?
} }
perms := perms(remote, user, event.Repo) perm, ok := perms_[event.Repo.FullName]
if perms != nil && perms.Pull { if !ok {
ws.WriteJSON(event) perm = perms(remote, user, event.Repo)
perms_[event.Repo.FullName] = perm
}
if perm != nil && perm.Pull {
err := ws.WriteJSON(event)
if err != nil {
log.Errorln(err, event)
}
} }
case <-ticker.C: case <-ticker.C:
ws.SetWriteDeadline(time.Now().Add(writeWait)) ws.SetWriteDeadline(time.Now().Add(writeWait))
err := ws.WriteMessage(websocket.PingMessage, []byte{}) err := ws.WriteMessage(websocket.PingMessage, []byte{})
if err != nil { if err != nil {
ws.Close() ws.Close()
log.Debugf("closed websocket")
return return
} }
} }
@ -72,6 +93,7 @@ func GetEvents(c *gin.Context) {
}() }()
readWebsocket(ws) readWebsocket(ws)
log.Debugf("closed websocket")
} }
// readWebsocket will block while reading the websocket data // readWebsocket will block while reading the websocket data