1
=== modified file 'rpc/jsoncodec/codec.go'
2
--- rpc/jsoncodec/codec.go	2013-05-28 07:01:54 +0000
3
+++ rpc/jsoncodec/codec.go	2013-06-07 17:45:32 +0000
4
@@ -8,6 +8,7 @@
5
8
	"launchpad.net/juju-core/log"
8
	"launchpad.net/juju-core/log"
6
9
	"launchpad.net/juju-core/rpc"
9
	"launchpad.net/juju-core/rpc"
7
10
	"sync"
10
	"sync"
8
11
	"sync/atomic"
9
11
)
12
)
10
12
13
11
13
// JSONConn sends and receives messages to an underlying connection
14
// JSONConn sends and receives messages to an underlying connection
12
@@ -20,26 +21,39 @@
13
20
	Close() error
21
	Close() error
14
21
}
22
}
15
22
23
20
23
var logRequests = true
24
// Codec implements rpc.Codec for a connection.
21
24
25
type Codec struct {
18
25
// codec implements rpc.Codec for a connection.
19
26
type codec struct {
22
27
	// msg holds the message that's just been read by ReadHeader, so
26
	// msg holds the message that's just been read by ReadHeader, so
23
28
	// that the body can be read by ReadBody.
27
	// that the body can be read by ReadBody.
28
29
	msg     inMsg
28
	msg         inMsg
29
30
	conn    JSONConn
29
	conn        JSONConn
30
31
	mu      sync.Mutex
30
	logMessages int32
31
32
	closing bool
31
	mu          sync.Mutex
32
32
	closing     bool
33
33
}
33
}
34
34
34
35
35
// New returns an rpc codec that uses conn to send and receive
35
// New returns an rpc codec that uses conn to send and receive
36
36
// messages.
36
// messages.
39
37
func New(conn JSONConn) rpc.Codec {
37
func New(conn JSONConn) *Codec {
40
38
	return &codec{
38
	return &Codec{
41
39
		conn: conn,
39
		conn: conn,
42
40
	}
40
	}
43
41
}
41
}
44
42
42
45
43
// SetLogging sets whether messages will be logged
46
44
// by the codec.
47
45
func (c *Codec) SetLogging(on bool) {
48
46
	val := int32(0)
49
47
	if on {
50
48
		val = 1
51
49
	}
52
50
	atomic.StoreInt32(&c.logMessages, val)
53
51
}
54
52
55
53
func (c *Codec) isLogging() bool {
56
54
	return atomic.LoadInt32(&c.logMessages) != 0
57
55
}
58
56
59
43
// inMsg holds an incoming message.  We don't know the type of the
57
// inMsg holds an incoming message.  We don't know the type of the
60
44
// parameters or response yet, so we delay parsing by storing them
58
// parameters or response yet, so we delay parsing by storing them
61
45
// in a RawMessage.
59
// in a RawMessage.
62
@@ -66,23 +80,23 @@
63
66
	Response  interface{} `json:",omitempty"`
80
	Response  interface{} `json:",omitempty"`
64
67
}
81
}
65
68
82
67
69
func (c *codec) Close() error {
83
func (c *Codec) Close() error {
68
70
	c.mu.Lock()
84
	c.mu.Lock()
69
71
	c.closing = true
85
	c.closing = true
70
72
	c.mu.Unlock()
86
	c.mu.Unlock()
71
73
	return c.conn.Close()
87
	return c.conn.Close()
72
74
}
88
}
73
75
89
75
76
func (c *codec) isClosing() bool {
90
func (c *Codec) isClosing() bool {
76
77
	c.mu.Lock()
91
	c.mu.Lock()
77
78
	defer c.mu.Unlock()
92
	defer c.mu.Unlock()
78
79
	return c.closing
93
	return c.closing
79
80
}
94
}
80
81
95
82
82
func (c *codec) ReadHeader(hdr *rpc.Header) error {
96
func (c *Codec) ReadHeader(hdr *rpc.Header) error {
83
83
	c.msg = inMsg{} // avoid any potential cross-message contamination.
97
	c.msg = inMsg{} // avoid any potential cross-message contamination.
84
84
	var err error
98
	var err error
86
85
	if logRequests {
99
	if c.isLogging() {
87
86
		var m json.RawMessage
100
		var m json.RawMessage
88
87
		err = c.conn.Receive(&m)
101
		err = c.conn.Receive(&m)
89
88
		if err == nil {
102
		if err == nil {
90
@@ -111,7 +125,7 @@
91
111
	return nil
125
	return nil
92
112
}
126
}
93
113
127
95
114
func (c *codec) ReadBody(body interface{}, isRequest bool) error {
128
func (c *Codec) ReadBody(body interface{}, isRequest bool) error {
96
115
	if body == nil {
129
	if body == nil {
97
116
		return nil
130
		return nil
98
117
	}
131
	}
99
@@ -129,7 +143,7 @@
100
129
	return json.Unmarshal(rawBody, body)
143
	return json.Unmarshal(rawBody, body)
101
130
}
144
}
102
131
145
104
132
func (c *codec) WriteMessage(hdr *rpc.Header, body interface{}) error {
146
func (c *Codec) WriteMessage(hdr *rpc.Header, body interface{}) error {
105
133
	r := &outMsg{
147
	r := &outMsg{
106
134
		RequestId: hdr.RequestId,
148
		RequestId: hdr.RequestId,
107
135
149
108
@@ -145,7 +159,7 @@
109
145
	} else {
159
	} else {
110
146
		r.Response = body
160
		r.Response = body
111
147
	}
161
	}
113
148
	if logRequests {
162
	if c.isLogging() {
114
149
		data, err := json.Marshal(r)
163
		data, err := json.Marshal(r)
115
150
		if err != nil {
164
		if err != nil {
116
151
			log.Debugf("rpc/jsoncodec: -> marshal error: %v", err)
165
			log.Debugf("rpc/jsoncodec: -> marshal error: %v", err)
117
152
166
118
=== modified file 'rpc/jsoncodec/codec_test.go'
119
--- rpc/jsoncodec/codec_test.go	2013-05-15 12:34:10 +0000
120
+++ rpc/jsoncodec/codec_test.go	2013-06-07 17:45:32 +0000
121
@@ -7,15 +7,19 @@
122
7
	. "launchpad.net/gocheck"
7
	. "launchpad.net/gocheck"
123
8
	"launchpad.net/juju-core/rpc"
8
	"launchpad.net/juju-core/rpc"
124
9
	"launchpad.net/juju-core/rpc/jsoncodec"
9
	"launchpad.net/juju-core/rpc/jsoncodec"
125
10
	"launchpad.net/juju-core/testing"
126
10
	"reflect"
11
	"reflect"
128
11
	"testing"
12
	"regexp"
129
13
	stdtesting "testing"
130
12
)
14
)
131
13
15
137
14
type suite struct{}
16
type suite struct {
138
15
17
	testing.LoggingSuite
139
16
var _ = Suite(suite{})
18
}
140
17
19
141
18
func TestPackage(t *testing.T) {
20
var _ = Suite(&suite{})
142
21
143
22
func TestPackage(t *stdtesting.T) {
144
19
	TestingT(t)
23
	TestingT(t)
145
20
}
24
}
146
21
25
147
@@ -52,7 +56,7 @@
148
52
	expectBody: &value{X: "result"},
56
	expectBody: &value{X: "result"},
149
53
}}
57
}}
150
54
58
152
55
func (suite) TestRead(c *C) {
59
func (*suite) TestRead(c *C) {
153
56
	for i, test := range readTests {
60
	for i, test := range readTests {
154
57
		c.Logf("test %d", i)
61
		c.Logf("test %d", i)
155
58
		codec := jsoncodec.New(&testConn{
62
		codec := jsoncodec.New(&testConn{
156
@@ -75,7 +79,97 @@
157
75
	}
79
	}
158
76
}
80
}
159
77
81
161
78
func (suite) TestErrorAfterClose(c *C) {
82
func (*suite) TestReadHeaderLogsRequests(c *C) {
162
83
	msg := `{"RequestId":1,"Type": "foo","Id": "id","Request":"frob","Params":{"X":"param"}}`
163
84
	codec := jsoncodec.New(&testConn{
164
85
		readMsgs: []string{msg, msg, msg},
165
86
	})
166
87
	// Check that logging is off by default
167
88
	var h rpc.Header
168
89
	err := codec.ReadHeader(&h)
169
90
	c.Assert(err, IsNil)
170
91
	c.Assert(c.GetTestLog(), Matches, "")
171
92
172
93
	// Check that we see a log message when we switch logging on.
173
94
	codec.SetLogging(true)
174
95
	err = codec.ReadHeader(&h)
175
96
	c.Assert(err, IsNil)
176
97
	c.Assert(c.GetTestLog(), Matches, ".*DEBUG rpc/jsoncodec: <- "+regexp.QuoteMeta(msg)+`\n`)
177
98
178
99
	// Check that we can switch it off again
179
100
	codec.SetLogging(false)
180
101
	err = codec.ReadHeader(&h)
181
102
	c.Assert(err, IsNil)
182
103
	c.Assert(c.GetTestLog(), Matches, ".*DEBUG rpc/jsoncodec: <- "+regexp.QuoteMeta(msg)+`\n`)
183
104
}
184
105
185
106
func (*suite) TestWriteMessageLogsRequests(c *C) {
186
107
	codec := jsoncodec.New(&testConn{})
187
108
	h := rpc.Header{
188
109
		RequestId: 1,
189
110
		Type:      "foo",
190
111
		Id:        "id",
191
112
		Request:   "frob",
192
113
	}
193
114
194
115
	// Check that logging is off by default
195
116
	err := codec.WriteMessage(&h, value{X: "param"})
196
117
	c.Assert(err, IsNil)
197
118
	c.Assert(c.GetTestLog(), Matches, "")
198
119
199
120
	// Check that we see a log message when we switch logging on.
200
121
	codec.SetLogging(true)
201
122
	err = codec.WriteMessage(&h, value{X: "param"})
202
123
	c.Assert(err, IsNil)
203
124
	msg := `{"RequestId":1,"Type":"foo","Id":"id","Request":"frob","Params":{"X":"param"}}`
204
125
	c.Assert(c.GetTestLog(), Matches, `.*DEBUG rpc/jsoncodec: -> `+regexp.QuoteMeta(msg)+`\n`)
205
126
206
127
	// Check that we can switch it off again
207
128
	codec.SetLogging(false)
208
129
	err = codec.WriteMessage(&h, value{X: "param"})
209
130
	c.Assert(err, IsNil)
210
131
	c.Assert(c.GetTestLog(), Matches, `.*DEBUG rpc/jsoncodec: -> `+regexp.QuoteMeta(msg)+`\n`)
211
132
}
212
133
213
134
func (*suite) TestConcurrentSetLoggingAndWrite(c *C) {
214
135
	// If log messages are not set atomically, this
215
136
	// test will fail when run under the race detector.
216
137
	codec := jsoncodec.New(&testConn{})
217
138
	done := make(chan struct{})
218
139
	go func() {
219
140
		codec.SetLogging(true)
220
141
		done <- struct{}{}
221
142
	}()
222
143
	h := rpc.Header{
223
144
		RequestId: 1,
224
145
		Type:      "foo",
225
146
		Id:        "id",
226
147
		Request:   "frob",
227
148
	}
228
149
	err := codec.WriteMessage(&h, value{X: "param"})
229
150
	c.Assert(err, IsNil)
230
151
	<-done
231
152
}
232
153
233
154
func (*suite) TestConcurrentSetLoggingAndRead(c *C) {
234
155
	// If log messages are not set atomically, this
235
156
	// test will fail when run under the race detector.
236
157
	msg := `{"RequestId":1,"Type": "foo","Id": "id","Request":"frob","Params":{"X":"param"}}`
237
158
	codec := jsoncodec.New(&testConn{
238
159
		readMsgs: []string{msg, msg, msg},
239
160
	})
240
161
	done := make(chan struct{})
241
162
	go func() {
242
163
		codec.SetLogging(true)
243
164
		done <- struct{}{}
244
165
	}()
245
166
	var h rpc.Header
246
167
	err := codec.ReadHeader(&h)
247
168
	c.Assert(err, IsNil)
248
169
	<-done
249
170
}
250
171
251
172
func (*suite) TestErrorAfterClose(c *C) {
252
79
	conn := &testConn{
173
	conn := &testConn{
253
80
		err: errors.New("some error"),
174
		err: errors.New("some error"),
254
81
	}
175
	}
255
@@ -105,7 +199,7 @@
256
105
		Request:   "frob",
199
		Request:   "frob",
257
106
	},
200
	},
258
107
	body:   &value{X: "param"},
201
	body:   &value{X: "param"},
260
108
	expect: `{"RequestId": 1, "Type": "foo", "Id": "id", "Request": "frob", "Params": {"X": "param"}}`,
202
	expect: `{"RequestId": 1, "Type": "foo","Id":"id", "Request": "frob", "Params": {"X": "param"}}`,
261
109
}, {
203
}, {
262
110
	hdr: &rpc.Header{
204
	hdr: &rpc.Header{
263
111
		RequestId: 2,
205
		RequestId: 2,
264
@@ -121,7 +215,7 @@
265
121
	expect: `{"RequestId": 3, "Response": {"X": "result"}}`,
215
	expect: `{"RequestId": 3, "Response": {"X": "result"}}`,
266
122
}}
216
}}
267
123
217
269
124
func (suite) TestWrite(c *C) {
218
func (*suite) TestWrite(c *C) {
270
125
	for i, test := range writeTests {
219
	for i, test := range writeTests {
271
126
		c.Logf("test %d", i)
220
		c.Logf("test %d", i)
272
127
		var conn testConn
221
		var conn testConn
273
128
222
274
=== modified file 'rpc/jsoncodec/conn.go'
275
--- rpc/jsoncodec/conn.go	2013-05-16 17:31:36 +0000
276
+++ rpc/jsoncodec/conn.go	2013-06-07 17:45:32 +0000
277
@@ -3,13 +3,12 @@
278
3
import (
3
import (
279
4
	"code.google.com/p/go.net/websocket"
4
	"code.google.com/p/go.net/websocket"
280
5
	"encoding/json"
5
	"encoding/json"
281
6
	"launchpad.net/juju-core/rpc"
282
7
	"net"
6
	"net"
283
8
)
7
)
284
9
8
285
10
// NewWebsocket returns an rpc codec that uses the given websocket
9
// NewWebsocket returns an rpc codec that uses the given websocket
286
11
// connection to send and receive messages.
10
// connection to send and receive messages.
288
12
func NewWebsocket(conn *websocket.Conn) rpc.Codec {
11
func NewWebsocket(conn *websocket.Conn) *Codec {
289
13
	return New(wsJSONConn{conn})
12
	return New(wsJSONConn{conn})
290
14
}
13
}
291
15
14
292
@@ -31,7 +30,7 @@
293
31
30
294
32
// NewNet returns an rpc codec that uses the given net
31
// NewNet returns an rpc codec that uses the given net
295
33
// connection to send and receive messages.
32
// connection to send and receive messages.
297
34
func NewNet(conn net.Conn) rpc.Codec {
33
func NewNet(conn net.Conn) *Codec {
298
35
	return New(&netConn{
34
	return New(&netConn{
299
36
		enc:  json.NewEncoder(conn),
35
		enc:  json.NewEncoder(conn),
300
37
		dec:  json.NewDecoder(conn),
36
		dec:  json.NewDecoder(conn),
301
38
37
302
=== modified file 'state/apiserver/apiserver.go'
303
--- state/apiserver/apiserver.go	2013-06-06 17:09:49 +0000
304
+++ state/apiserver/apiserver.go	2013-06-07 17:45:32 +0000
305
@@ -96,6 +96,7 @@
306
96
}
96
}
307
97
97
308
98
func (srv *Server) serveConn(wsConn *websocket.Conn) error {
98
func (srv *Server) serveConn(wsConn *websocket.Conn) error {
309
99
<<<<<<< TREE
310
99
	conn := rpc.NewConn(jsoncodec.NewWebsocket(wsConn))
100
	conn := rpc.NewConn(jsoncodec.NewWebsocket(wsConn))
311
100
	serverError := func(err error) error {
101
	serverError := func(err error) error {
312
101
		if err := common.ServerError(err); err != nil {
102
		if err := common.ServerError(err); err != nil {
313
@@ -103,6 +104,13 @@
314
103
		}
104
		}
315
104
		return nil
105
		return nil
316
105
	}
106
	}
317
107
=======
318
108
	codec := jsoncodec.NewWebsocket(wsConn)
319
109
	if log.Debug {
320
110
		codec.SetLogging(true)
321
111
	}
322
112
	conn := rpc.NewConn(codec)
323
113
>>>>>>> MERGE-SOURCE
324
106
	if err := conn.Serve(newStateServer(srv), serverError); err != nil {
114
	if err := conn.Serve(newStateServer(srv), serverError); err != nil {
325
107
		return err
115
		return err
326
108
	}
116
	}
Status:	Work in progress
Proposed branch:	lp:~rogpeppe/juju-core/313-jsoncodec-optional-log
Merge into:	lp:~juju/juju-core/trunk
Prerequisite:	lp:~rogpeppe/juju-core/312-api-jobs
Diff against target:	326 lines (+145/-30) (has conflicts) 4 files modified rpc/jsoncodec/codec.go (+31/-17) rpc/jsoncodec/codec_test.go (+104/-10) rpc/jsoncodec/conn.go (+2/-3) state/apiserver/apiserver.go (+8/-0) Text conflict in cmd/jujud/agent_test.go Text conflict in cmd/jujud/bootstrap_test.go Text conflict in environs/agent/agent.go Contents conflict in state/api/machine.go Text conflict in state/api/params/params.go Text conflict in state/apiserver/apiserver.go Contents conflict in state/apiserver/machine.go Contents conflict in state/apiserver/machine_test.go Text conflict in state/apiserver/utils.go
To merge this branch:	bzr merge lp:~rogpeppe/juju-core/313-jsoncodec-optional-log
Related bugs:	Link a bug report
Reviewer	Review Type	Date Requested	Status
Juju Engineering		2013-05-25	Pending
Review via email: mp+165736@code.launchpad.net