Mercurial > gemma

comparison pkg/auth/pool.go @ 493:8a0737aa6ab6 metamorph-for-all

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
The connection pool is now only a session store.
author Sascha L. Teichmann <sascha.teichmann@intevation.de>
date Fri, 24 Aug 2018 14:25:05 +0200
parents b2dc9c2f69e0
children
comparison
equal deleted inserted replaced
486:b2dc9c2f69e0 493:8a0737aa6ab6
6 "time" 6 "time"
7 7
8 bolt "github.com/coreos/bbolt" 8 bolt "github.com/coreos/bbolt"
9 ) 9 )
10 10
11 // ConnPool is the global connection pool. 11 // Sessions is the global connection pool.
12 var ConnPool *ConnectionPool 12 var Sessions *SessionStore
13 13
14 type ConnectionPool struct { 14 type SessionStore struct {
15 storage *bolt.DB 15 storage *bolt.DB
16 conns map[string]*Connection 16 conns map[string]*Connection
17 cmds chan func(*ConnectionPool) 17 cmds chan func(*SessionStore)
18 } 18 }
19 19
20 var sessionsBucket = []byte("sessions") 20 var sessionsBucket = []byte("sessions")
21 21
22 func NewConnectionPool(filename string) (*ConnectionPool, error) { 22 func NewSessionStore(filename string) (*SessionStore, error) {
23 23
24 pcp := &ConnectionPool{ 24 pcp := &SessionStore{
25 conns: make(map[string]*Connection), 25 conns: make(map[string]*Connection),
26 cmds: make(chan func(*ConnectionPool)), 26 cmds: make(chan func(*SessionStore)),
27 } 27 }
28 if err := pcp.openStorage(filename); err != nil { 28 if err := pcp.openStorage(filename); err != nil {
29 return nil, err 29 return nil, err
30 } 30 }
31 go pcp.run() 31 go pcp.run()
32 return pcp, nil 32 return pcp, nil
33 } 33 }
34 34
35 // openStorage opens a storage file. 35 // openStorage opens a storage file.
36 func (pcp *ConnectionPool) openStorage(filename string) error { 36 func (pcp *SessionStore) openStorage(filename string) error {
37 37
38 // No file, nothing to restore/persist. 38 // No file, nothing to restore/persist.
39 if filename == "" { 39 if filename == "" {
40 return nil 40 return nil
41 } 41 }
72 72
73 pcp.storage = db 73 pcp.storage = db
74 return nil 74 return nil
75 } 75 }
76 76
77 func (pcp *ConnectionPool) run() { 77 func (pcp *SessionStore) run() {
78 for { 78 for {
79 select { 79 select {
80 case cmd := <-pcp.cmds: 80 case cmd := <-pcp.cmds:
81 cmd(pcp) 81 cmd(pcp)
82 case <-time.After(time.Minute):
83 pcp.cleanDB()
84 case <-time.After(time.Minute * 5): 82 case <-time.After(time.Minute * 5):
85 pcp.cleanToken() 83 pcp.cleanToken()
86 } 84 }
87 } 85 }
88 } 86 }
89 87
90 func (pcp *ConnectionPool) cleanDB() { 88 func (pcp *SessionStore) cleanToken() {
91 }
92
93 func (pcp *ConnectionPool) cleanToken() {
94 now := time.Now() 89 now := time.Now()
95 for token, con := range pcp.conns { 90 for token, con := range pcp.conns {
96 expires := time.Unix(con.session.ExpiresAt, 0) 91 expires := time.Unix(con.session.ExpiresAt, 0)
97 if expires.Before(now) { 92 if expires.Before(now) {
98 // TODO: Be more graceful here?
99 con.close()
100 delete(pcp.conns, token) 93 delete(pcp.conns, token)
101 pcp.remove(token) 94 pcp.remove(token)
102 } 95 }
103 } 96 }
104 } 97 }
105 98
106 func (pcp *ConnectionPool) remove(token string) { 99 func (pcp *SessionStore) remove(token string) {
107 if pcp.storage == nil { 100 if pcp.storage == nil {
108 return 101 return
109 } 102 }
110 err := pcp.storage.Update(func(tx *bolt.Tx) error { 103 err := pcp.storage.Update(func(tx *bolt.Tx) error {
111 b := tx.Bucket(sessionsBucket) 104 b := tx.Bucket(sessionsBucket)
114 if err != nil { 107 if err != nil {
115 log.Printf("error: %v\n", err) 108 log.Printf("error: %v\n", err)
116 } 109 }
117 } 110 }
118 111
119 func (pcp *ConnectionPool) Delete(token string) bool { 112 func (pcp *SessionStore) Delete(token string) bool {
120 res := make(chan bool) 113 res := make(chan bool)
121 pcp.cmds <- func(pcp *ConnectionPool) { 114 pcp.cmds <- func(pcp *SessionStore) {
122 conn, found := pcp.conns[token] 115 if _, found := pcp.conns[token]; !found {
123 if !found {
124 res <- false 116 res <- false
125 return 117 return
126 } 118 }
127 conn.close()
128 delete(pcp.conns, token) 119 delete(pcp.conns, token)
129 pcp.remove(token) 120 pcp.remove(token)
130 res <- true 121 res <- true
131 } 122 }
132 return <-res 123 return <-res
133 } 124 }
134 125
135 func (pcp *ConnectionPool) store(token string, con *Connection) { 126 func (pcp *SessionStore) store(token string, con *Connection) {
136 if pcp.storage == nil { 127 if pcp.storage == nil {
137 return 128 return
138 } 129 }
139 err := pcp.storage.Update(func(tx *bolt.Tx) error { 130 err := pcp.storage.Update(func(tx *bolt.Tx) error {
140 b := tx.Bucket(sessionsBucket) 131 b := tx.Bucket(sessionsBucket)
147 if err != nil { 138 if err != nil {
148 log.Printf("error: %v\n", err) 139 log.Printf("error: %v\n", err)
149 } 140 }
150 } 141 }
151 142
152 func (pcp *ConnectionPool) Add(token string, session *Session) *Connection { 143 func (pcp *SessionStore) Add(token string, session *Session) *Connection {
153 res := make(chan *Connection) 144 res := make(chan *Connection)
154 145
155 pcp.cmds <- func(cp *ConnectionPool) { 146 pcp.cmds <- func(cp *SessionStore) {
156 con := pcp.conns[token] 147 con := pcp.conns[token]
157 if con == nil { 148 if con == nil {
158 con = &Connection{} 149 con = &Connection{}
159 pcp.conns[token] = con 150 pcp.conns[token] = con
160 } 151 }
165 156
166 con := <-res 157 con := <-res
167 return con 158 return con
168 } 159 }
169 160
170 func (pcp *ConnectionPool) Renew(token string) (string, error) { 161 func (pcp *SessionStore) Renew(token string) (string, error) {
171 162
172 type result struct { 163 type result struct {
173 newToken string 164 newToken string
174 err error 165 err error
175 } 166 }
176 167
177 resCh := make(chan result) 168 resCh := make(chan result)
178 169
179 pcp.cmds <- func(cp *ConnectionPool) { 170 pcp.cmds <- func(cp *SessionStore) {
180 con := pcp.conns[token] 171 con := pcp.conns[token]
181 if con == nil { 172 if con == nil {
182 resCh <- result{err: ErrNoSuchToken} 173 resCh <- result{err: ErrNoSuchToken}
183 } else { 174 } else {
184 delete(pcp.conns, token) 175 delete(pcp.conns, token)
194 185
195 r := <-resCh 186 r := <-resCh
196 return r.newToken, r.err 187 return r.newToken, r.err
197 } 188 }
198 189
199 func (pcp *ConnectionPool) Do(token string) (*Session, error) { 190 func (pcp *SessionStore) Do(token string) (*Session, error) {
200 191
201 type result struct { 192 type result struct {
202 session *Session 193 session *Session
203 err error 194 err error
204 } 195 }
205 196
206 res := make(chan result) 197 res := make(chan result)
207 198
208 pcp.cmds <- func(pcp *ConnectionPool) { 199 pcp.cmds <- func(pcp *SessionStore) {
209 con := pcp.conns[token] 200 con := pcp.conns[token]
210 if con == nil { 201 if con == nil {
211 res <- result{err: ErrNoSuchToken} 202 res <- result{err: ErrNoSuchToken}
212 return 203 return
213 } 204 }
224 } 215 }
225 216
226 return r.session, nil 217 return r.session, nil
227 } 218 }
228 219
229 func (pcp *ConnectionPool) Session(token string) *Session { 220 func (pcp *SessionStore) Session(token string) *Session {
230 res := make(chan *Session) 221 res := make(chan *Session)
231 pcp.cmds <- func(pcp *ConnectionPool) { 222 pcp.cmds <- func(pcp *SessionStore) {
232 con := pcp.conns[token] 223 con := pcp.conns[token]
233 if con == nil { 224 if con == nil {
234 res <- nil 225 res <- nil
235 } else { 226 } else {
236 con.touch() 227 con.touch()
239 } 230 }
240 } 231 }
241 return <-res 232 return <-res
242 } 233 }
243 234
244 func (pcp *ConnectionPool) Logout(user string) { 235 func (pcp *SessionStore) Logout(user string) {
245 pcp.cmds <- func(pcp *ConnectionPool) { 236 pcp.cmds <- func(pcp *SessionStore) {
246 for token, con := range pcp.conns { 237 for token, con := range pcp.conns {
247 if con.session.User == user { 238 if con.session.User == user {
248 delete(pcp.conns, token) 239 delete(pcp.conns, token)
249 pcp.remove(token) 240 pcp.remove(token)
250 } 241 }
251 } 242 }
252 } 243 }
253 } 244 }
254 245
255 func (pcp *ConnectionPool) Shutdown() error { 246 func (pcp *SessionStore) Shutdown() error {
256 if db := pcp.storage; db != nil { 247 if db := pcp.storage; db != nil {
257 log.Println("info: shutdown persistent connection pool.") 248 log.Println("info: shutdown persistent connection pool.")
258 pcp.storage = nil 249 pcp.storage = nil
259 return db.Close() 250 return db.Close()
260 } 251 }