Commit 320bc0fdd56ecc73f92518fbae04daf0bbe691ca - golang-github-go-kit-kit

examples/addsvc: update - Now a service interface, not just a single function - Proper clients for each transport - Latest and greatest transport/http helper(s) - Proper use of package log - gometalint clean \o/ Peter Bourgon 8 years ago

35 changed file(s) with 1351 addition(s) and 863 deletion(s). Raw diff Collapse all Expand all

-1

.gitignore less more

0	0	examples/addsvc/addsvc
1		examples/addsvc/client/addcli/addcli
	1	examples/addsvc/client/client
2	2	examples/stringsvc1/stringsvc1
3	3	examples/stringsvc2/stringsvc2
4	4	examples/stringsvc3/stringsvc3

-73

~~examples/addsvc/README.md~~ less more

0		# addsvc
1
2		addsvc is an example service, used to illustrate the mechanics of Go kit. It
3		exposes a single method to add two integers on a variety of transports and
4		endpoints.
5
6		## Highlights
7
8		### Configuration via flags
9
10		Go kit has no strong opinions about how to pass configuration to your service.
11		If your organization has established conventions to pass configuration into
12		your service, Go kit won't stand in your way. That said, package flag is a
13		good default: it's simple, well-understood, and provides a self-documenting
14		configuration surface area. Keeping with
15		[best practices](http://peter.bourgon.org/go-in-production/#configuration), flags
16		are defined in func main.
17
18		### Declarative composition
19
20		Go kit strongly favors explicit, declarative composition of interacting
21		components via a comprehensive func main. Time spent in keystrokes is made up
22		many, many times over when returning to the code and understanding exactly
23		what's happening, without having to unravel indirections or abstractions.
24
25		### Multiple transports
26
27		Go kit treats transports — HTTP, Thrift, gRPC, etc. — as pluggable. The same
28		service can be exposed on any, or multiple, available transports. The addsvc
29		example demonstrates how to make the same business logic available over
30		multiple transports simultaneously.
31
32		### Daemonizing
33
34		Go kit has no strong opinions about how to daemonize, supervise, or run your
35		service. If your organization has established conventions for running
36		services. Go kit won't stand in your way. Go kit services run equally well as
37		manually-copied binaries; applications provisioned with configuration
38		management tools like [Chef][], [Puppet][], or [Ansible][]; in containers like
39		[Docker][] or [rkt][]; or as part of a comprehensive scheduling platform like
40		[Kubernetes][], [Mesos][], [OpenStack][], [Deis][], etc.
41
42		[Chef]: https://www.chef.io
43		[Puppet]: https://puppetlabs.com
44		[Ansible]: http://www.ansible.com
45		[Docker]: http://docker.com
46		[rkt]: https://github.com/coreos/rkt
47		[Kubernetes]: http://kubernetes.io
48		[Mesos]: https://mesosphere.com
49		[OpenStack]: https://www.openstack.org
50		[Deis]: http://deis.io
51
52		## Server
53
54		To build and run addsvc,
55
56		```
57		go install
58		addsvc
59		```
60
61		## Client
62
63		addsvc comes with an example client, [addcli][].
64
65		[addcli]: https://github.com/go-kit/kit/blob/master/addsvc/client/addcli/main.go
66
67		```
68		$ cd client/addcli
69		$ go install
70		$ addcli
71		```
72

-2

examples/addsvc/_thrift/add.thrift less more

0		struct AddReply {
	0	struct SumReply {
1	1	1: i64 value
2	2	}
3	3
	4	struct ConcatReply {
	5	1: string value
	6	}
	7
4	8	service AddService {
5		AddReply Add(1: i64 a, 2: i64 b)
	9	SumReply Sum(1: i64 a, 2: i64 b)
	10	ConcatReply Concat(1: string a, 2: string b)
6	11	}

+21

-8

examples/addsvc/_thrift/gen-go/add/add_service-remote/add_service-remote.go less more

19	19	fmt.Fprintln(os.Stderr, "Usage of ", os.Args[0], " [-h host:port] [-u url] [-f[ramed]] function [arg1 [arg2...]]:")
20	20	flag.PrintDefaults()
21	21	fmt.Fprintln(os.Stderr, "\nFunctions:")
22		fmt.Fprintln(os.Stderr, " AddReply Add(i64 a, i64 b)")
	22	fmt.Fprintln(os.Stderr, " SumReply Sum(i64 a, i64 b)")
	23	fmt.Fprintln(os.Stderr, " ConcatReply Concat(string a, string b)")
23	24	fmt.Fprintln(os.Stderr)
24	25	os.Exit(0)
25	26	}

114	115	}
115	116
116	117	switch cmd {
117		case "Add":
	118	case "Sum":
118	119	if flag.NArg()-1 != 2 {
119		fmt.Fprintln(os.Stderr, "Add requires 2 args")
	120	fmt.Fprintln(os.Stderr, "Sum requires 2 args")
120	121	flag.Usage()
121	122	}
122		argvalue0, err4 := (strconv.ParseInt(flag.Arg(1), 10, 64))
123		if err4 != nil {
	123	argvalue0, err6 := (strconv.ParseInt(flag.Arg(1), 10, 64))
	124	if err6 != nil {
124	125	Usage()
125	126	return
126	127	}
127	128	value0 := argvalue0
128		argvalue1, err5 := (strconv.ParseInt(flag.Arg(2), 10, 64))
129		if err5 != nil {
	129	argvalue1, err7 := (strconv.ParseInt(flag.Arg(2), 10, 64))
	130	if err7 != nil {
130	131	Usage()
131	132	return
132	133	}
133	134	value1 := argvalue1
134		fmt.Print(client.Add(value0, value1))
	135	fmt.Print(client.Sum(value0, value1))
	136	fmt.Print("\n")
	137	break
	138	case "Concat":
	139	if flag.NArg()-1 != 2 {
	140	fmt.Fprintln(os.Stderr, "Concat requires 2 args")
	141	flag.Usage()
	142	}
	143	argvalue0 := flag.Arg(1)
	144	value0 := argvalue0
	145	argvalue1 := flag.Arg(2)
	146	value1 := argvalue1
	147	fmt.Print(client.Concat(value0, value1))
135	148	fmt.Print("\n")
136	149	break
137	150	case "":

+421

-78

examples/addsvc/_thrift/gen-go/add/addservice.go less more

5	5	import (
6	6	"bytes"
7	7	"fmt"
8
9	8	"github.com/apache/thrift/lib/go/thrift"
10	9	)
11	10

18	17	// Parameters:
19	18	// - A
20	19	// - B
21		Add(a int64, b int64) (r *AddReply, err error)
	20	Sum(a int64, b int64) (r *SumReply, err error)
	21	// Parameters:
	22	// - A
	23	// - B
	24	Concat(a string, b string) (r *ConcatReply, err error)
22	25	}
23	26
24	27	type AddServiceClient struct {

50	53	// Parameters:
51	54	// - A
52	55	// - B
53		func (p AddServiceClient) Add(a int64, b int64) (r AddReply, err error) {
54		if err = p.sendAdd(a, b); err != nil {
55		return
56		}
57		return p.recvAdd()
58		}
59
60		func (p *AddServiceClient) sendAdd(a int64, b int64) (err error) {
	56	func (p AddServiceClient) Sum(a int64, b int64) (r SumReply, err error) {
	57	if err = p.sendSum(a, b); err != nil {
	58	return
	59	}
	60	return p.recvSum()
	61	}
	62
	63	func (p *AddServiceClient) sendSum(a int64, b int64) (err error) {
61	64	oprot := p.OutputProtocol
62	65	if oprot == nil {
63	66	oprot = p.ProtocolFactory.GetProtocol(p.Transport)
64	67	p.OutputProtocol = oprot
65	68	}
66	69	p.SeqId++
67		if err = oprot.WriteMessageBegin("Add", thrift.CALL, p.SeqId); err != nil {
68		return
69		}
70		args := AddArgs{
	70	if err = oprot.WriteMessageBegin("Sum", thrift.CALL, p.SeqId); err != nil {
	71	return
	72	}
	73	args := SumArgs{
71	74	A: a,
72	75	B: b,
73	76	}

80	83	return oprot.Flush()
81	84	}
82	85
83		func (p AddServiceClient) recvAdd() (value AddReply, err error) {
	86	func (p AddServiceClient) recvSum() (value SumReply, err error) {
84	87	iprot := p.InputProtocol
85	88	if iprot == nil {
86	89	iprot = p.ProtocolFactory.GetProtocol(p.Transport)

104	107	return
105	108	}
106	109	if p.SeqId != seqId {
107		err = thrift.NewTApplicationException(thrift.BAD_SEQUENCE_ID, "Add failed: out of sequence response")
108		return
109		}
110		result := AddResult{}
	110	err = thrift.NewTApplicationException(thrift.BAD_SEQUENCE_ID, "Sum failed: out of sequence response")
	111	return
	112	}
	113	result := SumResult{}
	114	if err = result.Read(iprot); err != nil {
	115	return
	116	}
	117	if err = iprot.ReadMessageEnd(); err != nil {
	118	return
	119	}
	120	value = result.GetSuccess()
	121	return
	122	}
	123
	124	// Parameters:
	125	// - A
	126	// - B
	127	func (p AddServiceClient) Concat(a string, b string) (r ConcatReply, err error) {
	128	if err = p.sendConcat(a, b); err != nil {
	129	return
	130	}
	131	return p.recvConcat()
	132	}
	133
	134	func (p *AddServiceClient) sendConcat(a string, b string) (err error) {
	135	oprot := p.OutputProtocol
	136	if oprot == nil {
	137	oprot = p.ProtocolFactory.GetProtocol(p.Transport)
	138	p.OutputProtocol = oprot
	139	}
	140	p.SeqId++
	141	if err = oprot.WriteMessageBegin("Concat", thrift.CALL, p.SeqId); err != nil {
	142	return
	143	}
	144	args := ConcatArgs{
	145	A: a,
	146	B: b,
	147	}
	148	if err = args.Write(oprot); err != nil {
	149	return
	150	}
	151	if err = oprot.WriteMessageEnd(); err != nil {
	152	return
	153	}
	154	return oprot.Flush()
	155	}
	156
	157	func (p AddServiceClient) recvConcat() (value ConcatReply, err error) {
	158	iprot := p.InputProtocol
	159	if iprot == nil {
	160	iprot = p.ProtocolFactory.GetProtocol(p.Transport)
	161	p.InputProtocol = iprot
	162	}
	163	_, mTypeId, seqId, err := iprot.ReadMessageBegin()
	164	if err != nil {
	165	return
	166	}
	167	if mTypeId == thrift.EXCEPTION {
	168	error2 := thrift.NewTApplicationException(thrift.UNKNOWN_APPLICATION_EXCEPTION, "Unknown Exception")
	169	var error3 error
	170	error3, err = error2.Read(iprot)
	171	if err != nil {
	172	return
	173	}
	174	if err = iprot.ReadMessageEnd(); err != nil {
	175	return
	176	}
	177	err = error3
	178	return
	179	}
	180	if p.SeqId != seqId {
	181	err = thrift.NewTApplicationException(thrift.BAD_SEQUENCE_ID, "Concat failed: out of sequence response")
	182	return
	183	}
	184	result := ConcatResult{}
111	185	if err = result.Read(iprot); err != nil {
112	186	return
113	187	}

138	212
139	213	func NewAddServiceProcessor(handler AddService) *AddServiceProcessor {
140	214
141		self2 := &AddServiceProcessor{handler: handler, processorMap: make(map[string]thrift.TProcessorFunction)}
142		self2.processorMap["Add"] = &addServiceProcessorAdd{handler: handler}
143		return self2
	215	self4 := &AddServiceProcessor{handler: handler, processorMap: make(map[string]thrift.TProcessorFunction)}
	216	self4.processorMap["Sum"] = &addServiceProcessorSum{handler: handler}
	217	self4.processorMap["Concat"] = &addServiceProcessorConcat{handler: handler}
	218	return self4
144	219	}
145	220
146	221	func (p *AddServiceProcessor) Process(iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {

153	228	}
154	229	iprot.Skip(thrift.STRUCT)
155	230	iprot.ReadMessageEnd()
156		x3 := thrift.NewTApplicationException(thrift.UNKNOWN_METHOD, "Unknown function "+name)
	231	x5 := thrift.NewTApplicationException(thrift.UNKNOWN_METHOD, "Unknown function "+name)
157	232	oprot.WriteMessageBegin(name, thrift.EXCEPTION, seqId)
158		x3.Write(oprot)
	233	x5.Write(oprot)
159	234	oprot.WriteMessageEnd()
160	235	oprot.Flush()
161		return false, x3
162
163		}
164
165		type addServiceProcessorAdd struct {
	236	return false, x5
	237
	238	}
	239
	240	type addServiceProcessorSum struct {
166	241	handler AddService
167	242	}
168	243
169		func (p *addServiceProcessorAdd) Process(seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
170		args := AddArgs{}
	244	func (p *addServiceProcessorSum) Process(seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
	245	args := SumArgs{}
171	246	if err = args.Read(iprot); err != nil {
172	247	iprot.ReadMessageEnd()
173	248	x := thrift.NewTApplicationException(thrift.PROTOCOL_ERROR, err.Error())
174		oprot.WriteMessageBegin("Add", thrift.EXCEPTION, seqId)
	249	oprot.WriteMessageBegin("Sum", thrift.EXCEPTION, seqId)
175	250	x.Write(oprot)
176	251	oprot.WriteMessageEnd()
177	252	oprot.Flush()

179	254	}
180	255
181	256	iprot.ReadMessageEnd()
182		result := AddResult{}
183		var retval *AddReply
	257	result := SumResult{}
	258	var retval *SumReply
184	259	var err2 error
185		if retval, err2 = p.handler.Add(args.A, args.B); err2 != nil {
186		x := thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "Internal error processing Add: "+err2.Error())
187		oprot.WriteMessageBegin("Add", thrift.EXCEPTION, seqId)
	260	if retval, err2 = p.handler.Sum(args.A, args.B); err2 != nil {
	261	x := thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "Internal error processing Sum: "+err2.Error())
	262	oprot.WriteMessageBegin("Sum", thrift.EXCEPTION, seqId)
188	263	x.Write(oprot)
189	264	oprot.WriteMessageEnd()
190	265	oprot.Flush()

192	267	} else {
193	268	result.Success = retval
194	269	}
195		if err2 = oprot.WriteMessageBegin("Add", thrift.REPLY, seqId); err2 != nil {
	270	if err2 = oprot.WriteMessageBegin("Sum", thrift.REPLY, seqId); err2 != nil {
196	271	err = err2
197	272	}
198	273	if err2 = result.Write(oprot); err == nil && err2 != nil {

210	285	return true, err
211	286	}
212	287
	288	type addServiceProcessorConcat struct {
	289	handler AddService
	290	}
	291
	292	func (p *addServiceProcessorConcat) Process(seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
	293	args := ConcatArgs{}
	294	if err = args.Read(iprot); err != nil {
	295	iprot.ReadMessageEnd()
	296	x := thrift.NewTApplicationException(thrift.PROTOCOL_ERROR, err.Error())
	297	oprot.WriteMessageBegin("Concat", thrift.EXCEPTION, seqId)
	298	x.Write(oprot)
	299	oprot.WriteMessageEnd()
	300	oprot.Flush()
	301	return false, err
	302	}
	303
	304	iprot.ReadMessageEnd()
	305	result := ConcatResult{}
	306	var retval *ConcatReply
	307	var err2 error
	308	if retval, err2 = p.handler.Concat(args.A, args.B); err2 != nil {
	309	x := thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "Internal error processing Concat: "+err2.Error())
	310	oprot.WriteMessageBegin("Concat", thrift.EXCEPTION, seqId)
	311	x.Write(oprot)
	312	oprot.WriteMessageEnd()
	313	oprot.Flush()
	314	return true, err2
	315	} else {
	316	result.Success = retval
	317	}
	318	if err2 = oprot.WriteMessageBegin("Concat", thrift.REPLY, seqId); err2 != nil {
	319	err = err2
	320	}
	321	if err2 = result.Write(oprot); err == nil && err2 != nil {
	322	err = err2
	323	}
	324	if err2 = oprot.WriteMessageEnd(); err == nil && err2 != nil {
	325	err = err2
	326	}
	327	if err2 = oprot.Flush(); err == nil && err2 != nil {
	328	err = err2
	329	}
	330	if err != nil {
	331	return
	332	}
	333	return true, err
	334	}
	335
213	336	// HELPER FUNCTIONS AND STRUCTURES
214	337
215		type AddArgs struct {
	338	type SumArgs struct {
216	339	A int64 `thrift:"a,1" json:"a"`
217	340	B int64 `thrift:"b,2" json:"b"`
218	341	}
219	342
220		func NewAddArgs() *AddArgs {
221		return &AddArgs{}
222		}
223
224		func (p *AddArgs) GetA() int64 {
	343	func NewSumArgs() *SumArgs {
	344	return &SumArgs{}
	345	}
	346
	347	func (p *SumArgs) GetA() int64 {
225	348	return p.A
226	349	}
227	350
228		func (p *AddArgs) GetB() int64 {
	351	func (p *SumArgs) GetB() int64 {
229	352	return p.B
230	353	}
231		func (p *AddArgs) Read(iprot thrift.TProtocol) error {
	354	func (p *SumArgs) Read(iprot thrift.TProtocol) error {
232	355	if _, err := iprot.ReadStructBegin(); err != nil {
233	356	return fmt.Errorf("%T read error: %s", p, err)
234	357	}

264	387	return nil
265	388	}
266	389
267		func (p *AddArgs) ReadField1(iprot thrift.TProtocol) error {
	390	func (p *SumArgs) ReadField1(iprot thrift.TProtocol) error {
268	391	if v, err := iprot.ReadI64(); err != nil {
269	392	return fmt.Errorf("error reading field 1: %s", err)
270	393	} else {

273	396	return nil
274	397	}
275	398
276		func (p *AddArgs) ReadField2(iprot thrift.TProtocol) error {
	399	func (p *SumArgs) ReadField2(iprot thrift.TProtocol) error {
277	400	if v, err := iprot.ReadI64(); err != nil {
278	401	return fmt.Errorf("error reading field 2: %s", err)
279	402	} else {

282	405	return nil
283	406	}
284	407
285		func (p *AddArgs) Write(oprot thrift.TProtocol) error {
286		if err := oprot.WriteStructBegin("Add_args"); err != nil {
	408	func (p *SumArgs) Write(oprot thrift.TProtocol) error {
	409	if err := oprot.WriteStructBegin("Sum_args"); err != nil {
287	410	return fmt.Errorf("%T write struct begin error: %s", p, err)
288	411	}
289	412	if err := p.writeField1(oprot); err != nil {

301	424	return nil
302	425	}
303	426
304		func (p *AddArgs) writeField1(oprot thrift.TProtocol) (err error) {
	427	func (p *SumArgs) writeField1(oprot thrift.TProtocol) (err error) {
305	428	if err := oprot.WriteFieldBegin("a", thrift.I64, 1); err != nil {
306	429	return fmt.Errorf("%T write field begin error 1:a: %s", p, err)
307	430	}

314	437	return err
315	438	}
316	439
317		func (p *AddArgs) writeField2(oprot thrift.TProtocol) (err error) {
	440	func (p *SumArgs) writeField2(oprot thrift.TProtocol) (err error) {
318	441	if err := oprot.WriteFieldBegin("b", thrift.I64, 2); err != nil {
319	442	return fmt.Errorf("%T write field begin error 2:b: %s", p, err)
320	443	}

327	450	return err
328	451	}
329	452
330		func (p *AddArgs) String() string {
	453	func (p *SumArgs) String() string {
331	454	if p == nil {
332	455	return "<nil>"
333	456	}
334		return fmt.Sprintf("AddArgs(%+v)", *p)
335		}
336
337		type AddResult struct {
338		Success *AddReply `thrift:"success,0" json:"success"`
339		}
340
341		func NewAddResult() *AddResult {
342		return &AddResult{}
343		}
344
345		var AddResult_Success_DEFAULT *AddReply
346
347		func (p AddResult) GetSuccess() AddReply {
	457	return fmt.Sprintf("SumArgs(%+v)", *p)
	458	}
	459
	460	type SumResult struct {
	461	Success *SumReply `thrift:"success,0" json:"success"`
	462	}
	463
	464	func NewSumResult() *SumResult {
	465	return &SumResult{}
	466	}
	467
	468	var SumResult_Success_DEFAULT *SumReply
	469
	470	func (p SumResult) GetSuccess() SumReply {
348	471	if !p.IsSetSuccess() {
349		return AddResult_Success_DEFAULT
	472	return SumResult_Success_DEFAULT
350	473	}
351	474	return p.Success
352	475	}
353		func (p *AddResult) IsSetSuccess() bool {
	476	func (p *SumResult) IsSetSuccess() bool {
354	477	return p.Success != nil
355	478	}
356	479
357		func (p *AddResult) Read(iprot thrift.TProtocol) error {
	480	func (p *SumResult) Read(iprot thrift.TProtocol) error {
358	481	if _, err := iprot.ReadStructBegin(); err != nil {
359	482	return fmt.Errorf("%T read error: %s", p, err)
360	483	}

386	509	return nil
387	510	}
388	511
389		func (p *AddResult) ReadField0(iprot thrift.TProtocol) error {
390		p.Success = &AddReply{}
	512	func (p *SumResult) ReadField0(iprot thrift.TProtocol) error {
	513	p.Success = &SumReply{}
391	514	if err := p.Success.Read(iprot); err != nil {
392	515	return fmt.Errorf("%T error reading struct: %s", p.Success, err)
393	516	}
394	517	return nil
395	518	}
396	519
397		func (p *AddResult) Write(oprot thrift.TProtocol) error {
398		if err := oprot.WriteStructBegin("Add_result"); err != nil {
	520	func (p *SumResult) Write(oprot thrift.TProtocol) error {
	521	if err := oprot.WriteStructBegin("Sum_result"); err != nil {
399	522	return fmt.Errorf("%T write struct begin error: %s", p, err)
400	523	}
401	524	if err := p.writeField0(oprot); err != nil {

410	533	return nil
411	534	}
412	535
413		func (p *AddResult) writeField0(oprot thrift.TProtocol) (err error) {
	536	func (p *SumResult) writeField0(oprot thrift.TProtocol) (err error) {
414	537	if p.IsSetSuccess() {
415	538	if err := oprot.WriteFieldBegin("success", thrift.STRUCT, 0); err != nil {
416	539	return fmt.Errorf("%T write field begin error 0:success: %s", p, err)

425	548	return err
426	549	}
427	550
428		func (p *AddResult) String() string {
	551	func (p *SumResult) String() string {
429	552	if p == nil {
430	553	return "<nil>"
431	554	}
432		return fmt.Sprintf("AddResult(%+v)", *p)
433		}
	555	return fmt.Sprintf("SumResult(%+v)", *p)
	556	}
	557
	558	type ConcatArgs struct {
	559	A string `thrift:"a,1" json:"a"`
	560	B string `thrift:"b,2" json:"b"`
	561	}
	562
	563	func NewConcatArgs() *ConcatArgs {
	564	return &ConcatArgs{}
	565	}
	566
	567	func (p *ConcatArgs) GetA() string {
	568	return p.A
	569	}
	570
	571	func (p *ConcatArgs) GetB() string {
	572	return p.B
	573	}
	574	func (p *ConcatArgs) Read(iprot thrift.TProtocol) error {
	575	if _, err := iprot.ReadStructBegin(); err != nil {
	576	return fmt.Errorf("%T read error: %s", p, err)
	577	}
	578	for {
	579	_, fieldTypeId, fieldId, err := iprot.ReadFieldBegin()
	580	if err != nil {
	581	return fmt.Errorf("%T field %d read error: %s", p, fieldId, err)
	582	}
	583	if fieldTypeId == thrift.STOP {
	584	break
	585	}
	586	switch fieldId {
	587	case 1:
	588	if err := p.ReadField1(iprot); err != nil {
	589	return err
	590	}
	591	case 2:
	592	if err := p.ReadField2(iprot); err != nil {
	593	return err
	594	}
	595	default:
	596	if err := iprot.Skip(fieldTypeId); err != nil {
	597	return err
	598	}
	599	}
	600	if err := iprot.ReadFieldEnd(); err != nil {
	601	return err
	602	}
	603	}
	604	if err := iprot.ReadStructEnd(); err != nil {
	605	return fmt.Errorf("%T read struct end error: %s", p, err)
	606	}
	607	return nil
	608	}
	609
	610	func (p *ConcatArgs) ReadField1(iprot thrift.TProtocol) error {
	611	if v, err := iprot.ReadString(); err != nil {
	612	return fmt.Errorf("error reading field 1: %s", err)
	613	} else {
	614	p.A = v
	615	}
	616	return nil
	617	}
	618
	619	func (p *ConcatArgs) ReadField2(iprot thrift.TProtocol) error {
	620	if v, err := iprot.ReadString(); err != nil {
	621	return fmt.Errorf("error reading field 2: %s", err)
	622	} else {
	623	p.B = v
	624	}
	625	return nil
	626	}
	627
	628	func (p *ConcatArgs) Write(oprot thrift.TProtocol) error {
	629	if err := oprot.WriteStructBegin("Concat_args"); err != nil {
	630	return fmt.Errorf("%T write struct begin error: %s", p, err)
	631	}
	632	if err := p.writeField1(oprot); err != nil {
	633	return err
	634	}
	635	if err := p.writeField2(oprot); err != nil {
	636	return err
	637	}
	638	if err := oprot.WriteFieldStop(); err != nil {
	639	return fmt.Errorf("write field stop error: %s", err)
	640	}
	641	if err := oprot.WriteStructEnd(); err != nil {
	642	return fmt.Errorf("write struct stop error: %s", err)
	643	}
	644	return nil
	645	}
	646
	647	func (p *ConcatArgs) writeField1(oprot thrift.TProtocol) (err error) {
	648	if err := oprot.WriteFieldBegin("a", thrift.STRING, 1); err != nil {
	649	return fmt.Errorf("%T write field begin error 1:a: %s", p, err)
	650	}
	651	if err := oprot.WriteString(string(p.A)); err != nil {
	652	return fmt.Errorf("%T.a (1) field write error: %s", p, err)
	653	}
	654	if err := oprot.WriteFieldEnd(); err != nil {
	655	return fmt.Errorf("%T write field end error 1:a: %s", p, err)
	656	}
	657	return err
	658	}
	659
	660	func (p *ConcatArgs) writeField2(oprot thrift.TProtocol) (err error) {
	661	if err := oprot.WriteFieldBegin("b", thrift.STRING, 2); err != nil {
	662	return fmt.Errorf("%T write field begin error 2:b: %s", p, err)
	663	}
	664	if err := oprot.WriteString(string(p.B)); err != nil {
	665	return fmt.Errorf("%T.b (2) field write error: %s", p, err)
	666	}
	667	if err := oprot.WriteFieldEnd(); err != nil {
	668	return fmt.Errorf("%T write field end error 2:b: %s", p, err)
	669	}
	670	return err
	671	}
	672
	673	func (p *ConcatArgs) String() string {
	674	if p == nil {
	675	return "<nil>"
	676	}
	677	return fmt.Sprintf("ConcatArgs(%+v)", *p)
	678	}
	679
	680	type ConcatResult struct {
	681	Success *ConcatReply `thrift:"success,0" json:"success"`
	682	}
	683
	684	func NewConcatResult() *ConcatResult {
	685	return &ConcatResult{}
	686	}
	687
	688	var ConcatResult_Success_DEFAULT *ConcatReply
	689
	690	func (p ConcatResult) GetSuccess() ConcatReply {
	691	if !p.IsSetSuccess() {
	692	return ConcatResult_Success_DEFAULT
	693	}
	694	return p.Success
	695	}
	696	func (p *ConcatResult) IsSetSuccess() bool {
	697	return p.Success != nil
	698	}
	699
	700	func (p *ConcatResult) Read(iprot thrift.TProtocol) error {
	701	if _, err := iprot.ReadStructBegin(); err != nil {
	702	return fmt.Errorf("%T read error: %s", p, err)
	703	}
	704	for {
	705	_, fieldTypeId, fieldId, err := iprot.ReadFieldBegin()
	706	if err != nil {
	707	return fmt.Errorf("%T field %d read error: %s", p, fieldId, err)
	708	}
	709	if fieldTypeId == thrift.STOP {
	710	break
	711	}
	712	switch fieldId {
	713	case 0:
	714	if err := p.ReadField0(iprot); err != nil {
	715	return err
	716	}
	717	default:
	718	if err := iprot.Skip(fieldTypeId); err != nil {
	719	return err
	720	}
	721	}
	722	if err := iprot.ReadFieldEnd(); err != nil {
	723	return err
	724	}
	725	}
	726	if err := iprot.ReadStructEnd(); err != nil {
	727	return fmt.Errorf("%T read struct end error: %s", p, err)
	728	}
	729	return nil
	730	}
	731
	732	func (p *ConcatResult) ReadField0(iprot thrift.TProtocol) error {
	733	p.Success = &ConcatReply{}
	734	if err := p.Success.Read(iprot); err != nil {
	735	return fmt.Errorf("%T error reading struct: %s", p.Success, err)
	736	}
	737	return nil
	738	}
	739
	740	func (p *ConcatResult) Write(oprot thrift.TProtocol) error {
	741	if err := oprot.WriteStructBegin("Concat_result"); err != nil {
	742	return fmt.Errorf("%T write struct begin error: %s", p, err)
	743	}
	744	if err := p.writeField0(oprot); err != nil {
	745	return err
	746	}
	747	if err := oprot.WriteFieldStop(); err != nil {
	748	return fmt.Errorf("write field stop error: %s", err)
	749	}
	750	if err := oprot.WriteStructEnd(); err != nil {
	751	return fmt.Errorf("write struct stop error: %s", err)
	752	}
	753	return nil
	754	}
	755
	756	func (p *ConcatResult) writeField0(oprot thrift.TProtocol) (err error) {
	757	if p.IsSetSuccess() {
	758	if err := oprot.WriteFieldBegin("success", thrift.STRUCT, 0); err != nil {
	759	return fmt.Errorf("%T write field begin error 0:success: %s", p, err)
	760	}
	761	if err := p.Success.Write(oprot); err != nil {
	762	return fmt.Errorf("%T error writing struct: %s", p.Success, err)
	763	}
	764	if err := oprot.WriteFieldEnd(); err != nil {
	765	return fmt.Errorf("%T write field end error 0:success: %s", p, err)
	766	}
	767	}
	768	return err
	769	}
	770
	771	func (p *ConcatResult) String() string {
	772	if p == nil {
	773	return "<nil>"
	774	}
	775	return fmt.Sprintf("ConcatResult(%+v)", *p)
	776	}

+99

-11

examples/addsvc/_thrift/gen-go/add/ttypes.go less more

15	15
16	16	var GoUnusedProtection__ int
17	17
18		type AddReply struct {
	18	type SumReply struct {
19	19	Value int64 `thrift:"value,1" json:"value"`
20	20	}
21	21
22		func NewAddReply() *AddReply {
23		return &AddReply{}
	22	func NewSumReply() *SumReply {
	23	return &SumReply{}
24	24	}
25	25
26		func (p *AddReply) GetValue() int64 {
	26	func (p *SumReply) GetValue() int64 {
27	27	return p.Value
28	28	}
29		func (p *AddReply) Read(iprot thrift.TProtocol) error {
	29	func (p *SumReply) Read(iprot thrift.TProtocol) error {
30	30	if _, err := iprot.ReadStructBegin(); err != nil {
31	31	return fmt.Errorf("%T read error: %s", p, err)
32	32	}

58	58	return nil
59	59	}
60	60
61		func (p *AddReply) ReadField1(iprot thrift.TProtocol) error {
	61	func (p *SumReply) ReadField1(iprot thrift.TProtocol) error {
62	62	if v, err := iprot.ReadI64(); err != nil {
63	63	return fmt.Errorf("error reading field 1: %s", err)
64	64	} else {

67	67	return nil
68	68	}
69	69
70		func (p *AddReply) Write(oprot thrift.TProtocol) error {
71		if err := oprot.WriteStructBegin("AddReply"); err != nil {
	70	func (p *SumReply) Write(oprot thrift.TProtocol) error {
	71	if err := oprot.WriteStructBegin("SumReply"); err != nil {
72	72	return fmt.Errorf("%T write struct begin error: %s", p, err)
73	73	}
74	74	if err := p.writeField1(oprot); err != nil {

83	83	return nil
84	84	}
85	85
86		func (p *AddReply) writeField1(oprot thrift.TProtocol) (err error) {
	86	func (p *SumReply) writeField1(oprot thrift.TProtocol) (err error) {
87	87	if err := oprot.WriteFieldBegin("value", thrift.I64, 1); err != nil {
88	88	return fmt.Errorf("%T write field begin error 1:value: %s", p, err)
89	89	}

96	96	return err
97	97	}
98	98
99		func (p *AddReply) String() string {
	99	func (p *SumReply) String() string {
100	100	if p == nil {
101	101	return "<nil>"
102	102	}
103		return fmt.Sprintf("AddReply(%+v)", *p)
	103	return fmt.Sprintf("SumReply(%+v)", *p)
104	104	}
	105
	106	type ConcatReply struct {
	107	Value string `thrift:"value,1" json:"value"`
	108	}
	109
	110	func NewConcatReply() *ConcatReply {
	111	return &ConcatReply{}
	112	}
	113
	114	func (p *ConcatReply) GetValue() string {
	115	return p.Value
	116	}
	117	func (p *ConcatReply) Read(iprot thrift.TProtocol) error {
	118	if _, err := iprot.ReadStructBegin(); err != nil {
	119	return fmt.Errorf("%T read error: %s", p, err)
	120	}
	121	for {
	122	_, fieldTypeId, fieldId, err := iprot.ReadFieldBegin()
	123	if err != nil {
	124	return fmt.Errorf("%T field %d read error: %s", p, fieldId, err)
	125	}
	126	if fieldTypeId == thrift.STOP {
	127	break
	128	}
	129	switch fieldId {
	130	case 1:
	131	if err := p.ReadField1(iprot); err != nil {
	132	return err
	133	}
	134	default:
	135	if err := iprot.Skip(fieldTypeId); err != nil {
	136	return err
	137	}
	138	}
	139	if err := iprot.ReadFieldEnd(); err != nil {
	140	return err
	141	}
	142	}
	143	if err := iprot.ReadStructEnd(); err != nil {
	144	return fmt.Errorf("%T read struct end error: %s", p, err)
	145	}
	146	return nil
	147	}
	148
	149	func (p *ConcatReply) ReadField1(iprot thrift.TProtocol) error {
	150	if v, err := iprot.ReadString(); err != nil {
	151	return fmt.Errorf("error reading field 1: %s", err)
	152	} else {
	153	p.Value = v
	154	}
	155	return nil
	156	}
	157
	158	func (p *ConcatReply) Write(oprot thrift.TProtocol) error {
	159	if err := oprot.WriteStructBegin("ConcatReply"); err != nil {
	160	return fmt.Errorf("%T write struct begin error: %s", p, err)
	161	}
	162	if err := p.writeField1(oprot); err != nil {
	163	return err
	164	}
	165	if err := oprot.WriteFieldStop(); err != nil {
	166	return fmt.Errorf("write field stop error: %s", err)
	167	}
	168	if err := oprot.WriteStructEnd(); err != nil {
	169	return fmt.Errorf("write struct stop error: %s", err)
	170	}
	171	return nil
	172	}
	173
	174	func (p *ConcatReply) writeField1(oprot thrift.TProtocol) (err error) {
	175	if err := oprot.WriteFieldBegin("value", thrift.STRING, 1); err != nil {
	176	return fmt.Errorf("%T write field begin error 1:value: %s", p, err)
	177	}
	178	if err := oprot.WriteString(string(p.Value)); err != nil {
	179	return fmt.Errorf("%T.value (1) field write error: %s", p, err)
	180	}
	181	if err := oprot.WriteFieldEnd(); err != nil {
	182	return fmt.Errorf("%T write field end error 1:value: %s", p, err)
	183	}
	184	return err
	185	}
	186
	187	func (p *ConcatReply) String() string {
	188	if p == nil {
	189	return "<nil>"
	190	}
	191	return fmt.Sprintf("ConcatReply(%+v)", *p)
	192	}

-35

~~examples/addsvc/add.go~~ less more

0		package main
1
2		import (
3		"golang.org/x/net/context"
4
5		"github.com/go-kit/kit/endpoint"
6		"github.com/go-kit/kit/examples/addsvc/reqrep"
7		"github.com/go-kit/kit/log"
8		)
9
10		// Add is the abstract definition of what this service does. It could easily
11		// be an interface type with multiple methods, in which case each method would
12		// be an endpoint.
13		type Add func(context.Context, int64, int64) int64
14
15		// pureAdd implements Add with no dependencies.
16		func pureAdd(_ context.Context, a, b int64) int64 { return a + b }
17
18		// proxyAdd returns an implementation of Add that invokes a remote Add
19		// service.
20		func proxyAdd(remote endpoint.Endpoint, logger log.Logger) Add {
21		return func(ctx context.Context, a, b int64) int64 {
22		resp, err := remote(ctx, reqrep.AddRequest{A: a, B: b})
23		if err != nil {
24		logger.Log("err", err)
25		return 0
26		}
27		addResp, ok := resp.(reqrep.AddResponse)
28		if !ok {
29		logger.Log("err", endpoint.ErrBadCast)
30		return 0
31		}
32		return addResp.V
33		}
34		}

-126

~~examples/addsvc/client/addcli/main.go~~ less more

0		package main
1
2		import (
3		"flag"
4		"log"
5		"net/rpc"
6		"net/url"
7		"os"
8		"strings"
9
10		"golang.org/x/net/context"
11		"google.golang.org/grpc"
12
13		"github.com/apache/thrift/lib/go/thrift"
14		"github.com/go-kit/kit/endpoint"
15		thriftadd "github.com/go-kit/kit/examples/addsvc/_thrift/gen-go/add"
16		grpcclient "github.com/go-kit/kit/examples/addsvc/client/grpc"
17		httpclient "github.com/go-kit/kit/examples/addsvc/client/http"
18		netrpcclient "github.com/go-kit/kit/examples/addsvc/client/netrpc"
19		thriftclient "github.com/go-kit/kit/examples/addsvc/client/thrift"
20		"github.com/go-kit/kit/examples/addsvc/reqrep"
21		)
22
23		func main() {
24		// Flag domain. Note that gRPC transitively registers flags via its import
25		// of glog. So, we define a new flag set, to keep those domains distinct.
26		fs := flag.NewFlagSet("", flag.ExitOnError)
27		var (
28		transport = fs.String("transport", "grpc", "http, grpc, netrpc, thrift")
29		httpAddr = fs.String("http.addr", "localhost:8001", "HTTP (JSON) address")
30		grpcAddr = fs.String("grpc.addr", "localhost:8002", "gRPC address")
31		netrpcAddr = fs.String("netrpc.addr", "localhost:8003", "net/rpc address")
32		thriftAddr = fs.String("thrift.addr", "localhost:8004", "Thrift address")
33		thriftProtocol = fs.String("thrift.protocol", "binary", "binary, compact, json, simplejson")
34		thriftBufferSize = fs.Int("thrift.buffer.size", 0, "0 for unbuffered")
35		thriftFramed = fs.Bool("thrift.framed", false, "true to enable framing")
36		a = fs.Int64("a", 1, "a value")
37		b = fs.Int64("b", 2, "b value")
38		)
39		flag.Usage = fs.Usage // only show our flags
40		fs.Parse(os.Args[1:])
41		log.SetFlags(0)
42		log.SetOutput(os.Stdout)
43
44		var e endpoint.Endpoint
45		switch *transport {
46		case "http":
47		if !strings.HasPrefix(*httpAddr, "http") {
48		httpAddr = "http://" + httpAddr
49		}
50		u, err := url.Parse(*httpAddr)
51		if err != nil {
52		log.Fatalf("url.Parse: %v", err)
53		}
54		if u.Path == "" {
55		u.Path = "/add"
56		}
57		e = httpclient.NewClient("GET", u.String())
58
59		case "grpc":
60		cc, err := grpc.Dial(*grpcAddr)
61		if err != nil {
62		log.Fatalf("grpc.Dial: %v", err)
63		}
64		e = grpcclient.NewClient(cc)
65
66		case "netrpc":
67		client, err := rpc.DialHTTP("tcp", *netrpcAddr)
68		if err != nil {
69		log.Fatalf("rpc.DialHTTP: %v", err)
70		}
71		e = netrpcclient.NewClient(client)
72
73		case "thrift":
74		var protocolFactory thrift.TProtocolFactory
75		switch *thriftProtocol {
76		case "compact":
77		protocolFactory = thrift.NewTCompactProtocolFactory()
78		case "simplejson":
79		protocolFactory = thrift.NewTSimpleJSONProtocolFactory()
80		case "json":
81		protocolFactory = thrift.NewTJSONProtocolFactory()
82		case "binary", "":
83		protocolFactory = thrift.NewTBinaryProtocolFactoryDefault()
84		default:
85		log.Fatalf("invalid protocol %q", *thriftProtocol)
86		}
87
88		var transportFactory thrift.TTransportFactory
89		if *thriftBufferSize > 0 {
90		transportFactory = thrift.NewTBufferedTransportFactory(*thriftBufferSize)
91		} else {
92		transportFactory = thrift.NewTTransportFactory()
93		}
94
95		if *thriftFramed {
96		transportFactory = thrift.NewTFramedTransportFactory(transportFactory)
97		}
98
99		transportSocket, err := thrift.NewTSocket(*thriftAddr)
100		if err != nil {
101		log.Fatalf("thrift.NewTSocket: %v", err)
102		}
103
104		transport := transportFactory.GetTransport(transportSocket)
105		defer transport.Close()
106		if err := transport.Open(); err != nil {
107		log.Fatalf("Thrift transport.Open: %v", err)
108		}
109
110		e = thriftclient.NewClient(thriftadd.NewAddServiceClientFactory(transport, protocolFactory))
111
112		default:
113		log.Fatalf("unsupported transport %q", *transport)
114		}
115
116		response, err := e(context.Background(), reqrep.AddRequest{A: a, B: b})
117		if err != nil {
118		log.Fatalf("when invoking request: %v", err)
119		}
120		addResponse, ok := response.(reqrep.AddResponse)
121		if !ok {
122		log.Fatalf("when type-asserting response: %v", endpoint.ErrBadCast)
123		}
124		log.Print(addResponse.V)
125		}

+51

-32

examples/addsvc/client/grpc/client.go less more

0	0	package grpc
1	1
2	2	import (
	3	"github.com/go-kit/kit/log"
3	4	"golang.org/x/net/context"
4	5	"google.golang.org/grpc"
5	6
6		"github.com/go-kit/kit/endpoint"
7	7	"github.com/go-kit/kit/examples/addsvc/pb"
8		"github.com/go-kit/kit/examples/addsvc/reqrep"
	8	"github.com/go-kit/kit/examples/addsvc/server"
9	9	)
10	10
11		// NewClient takes a gRPC ClientConn that should point to an instance of an
12		// addsvc. It returns an endpoint that wraps and invokes that ClientConn.
13		func NewClient(cc *grpc.ClientConn) endpoint.Endpoint {
14		client := pb.NewAddClient(cc)
15		return func(ctx context.Context, request interface{}) (interface{}, error) {
16		var (
17		errs = make(chan error, 1)
18		responses = make(chan interface{}, 1)
19		)
20		go func() {
21		addReq, ok := request.(reqrep.AddRequest)
22		if !ok {
23		errs <- endpoint.ErrBadCast
24		return
25		}
26		reply, err := client.Add(ctx, &pb.AddRequest{A: addReq.A, B: addReq.B})
27		if err != nil {
28		errs <- err
29		return
30		}
31		responses <- reqrep.AddResponse{V: reply.V}
32		}()
33		select {
34		case <-ctx.Done():
35		return nil, context.DeadlineExceeded
36		case err := <-errs:
37		return nil, err
38		case response := <-responses:
39		return response, nil
40		}
	11	// New returns an AddService that's backed by the provided ClientConn.
	12	func New(ctx context.Context, cc *grpc.ClientConn, logger log.Logger) server.AddService {
	13	return client{ctx, pb.NewAddClient(cc), logger}
	14	}
	15
	16	type client struct {
	17	context.Context
	18	pb.AddClient
	19	log.Logger
	20	}
	21
	22	// TODO(pb): If your service interface methods don't return an error, we have
	23	// no way to signal problems with a service client. If they don't take a
	24	// context, we have to provide a global context for any transport that
	25	// requires one, effectively making your service a black box to any context-
	26	// specific information. So, we should make some recommendations:
	27	//
	28	// - To get started, a simple service interface is probably fine.
	29	//
	30	// - To properly deal with transport errors, every method on your service
	31	// should return an error. This is probably important.
	32	//
	33	// - To properly deal with context information, every method on your service
	34	// can take a context as its first argument. This may or may not be
	35	// important.
	36
	37	func (c client) Sum(a, b int) int {
	38	request := &pb.SumRequest{
	39	A: int64(a),
	40	B: int64(b),
41	41	}
	42	reply, err := c.AddClient.Sum(c.Context, request)
	43	if err != nil {
	44	_ = c.Logger.Log("err", err) // Without an error return parameter, we can't do anything else...
	45	return 0
	46	}
	47	return int(reply.V)
42	48	}
	49
	50	func (c client) Concat(a, b string) string {
	51	request := &pb.ConcatRequest{
	52	A: a,
	53	B: b,
	54	}
	55	reply, err := c.AddClient.Concat(c.Context, request)
	56	if err != nil {
	57	_ = c.Logger.Log("err", err)
	58	return ""
	59	}
	60	return reply.V
	61	}

-61

~~examples/addsvc/client/http/client.go~~ less more

0		package http
1
2		import (
3		"bytes"
4		"encoding/json"
5		"net/http"
6
7		"golang.org/x/net/context"
8
9		"github.com/go-kit/kit/endpoint"
10		"github.com/go-kit/kit/examples/addsvc/reqrep"
11		httptransport "github.com/go-kit/kit/transport/http"
12		)
13
14		// NewClient takes a URL that should point to an instance of an addsvc. It
15		// returns an endpoint that makes a request to that URL.
16		func NewClient(method, url string, before ...httptransport.RequestFunc) endpoint.Endpoint {
17		return func(ctx0 context.Context, request interface{}) (interface{}, error) {
18		var (
19		ctx, cancel = context.WithCancel(ctx0)
20		errs = make(chan error, 1)
21		responses = make(chan interface{}, 1)
22		)
23		defer cancel()
24		go func() {
25		var buf bytes.Buffer
26		if err := json.NewEncoder(&buf).Encode(request); err != nil {
27		errs <- err
28		return
29		}
30		req, err := http.NewRequest(method, url, &buf)
31		if err != nil {
32		errs <- err
33		return
34		}
35		for _, f := range before {
36		ctx = f(ctx, req)
37		}
38		resp, err := http.DefaultClient.Do(req)
39		if err != nil {
40		errs <- err
41		return
42		}
43		defer resp.Body.Close()
44		var response reqrep.AddResponse
45		if err := json.NewDecoder(resp.Body).Decode(&response); err != nil {
46		errs <- err
47		return
48		}
49		responses <- response
50		}()
51		select {
52		case <-ctx.Done():
53		return nil, context.DeadlineExceeded
54		case err := <-errs:
55		return nil, err
56		case response := <-responses:
57		return response, nil
58		}
59		}
60		}

+72

-0

examples/addsvc/client/httpjson/client.go less more

	0	package httpjson
	1
	2	import (
	3	"net/http"
	4	"net/url"
	5
	6	"golang.org/x/net/context"
	7
	8	"github.com/go-kit/kit/endpoint"
	9	"github.com/go-kit/kit/examples/addsvc/server"
	10	"github.com/go-kit/kit/log"
	11	httptransport "github.com/go-kit/kit/transport/http"
	12	)
	13
	14	// New returns an AddService that's backed by the URL. baseurl will have its
	15	// scheme and hostport used, but its path will be overwritten. If client is
	16	// nil, http.DefaultClient will be used.
	17	func New(ctx context.Context, baseurl url.URL, logger log.Logger, c http.Client) server.AddService {
	18	sumURL, err := url.Parse(baseurl.String())
	19	if err != nil {
	20	panic(err)
	21	}
	22	concatURL, err := url.Parse(baseurl.String())
	23	if err != nil {
	24	panic(err)
	25	}
	26	sumURL.Path = "/sum"
	27	concatURL.Path = "/concat"
	28	return client{
	29	Context: ctx,
	30	Logger: logger,
	31	sum: (httptransport.Client{
	32	Client: c,
	33	Method: "GET",
	34	URL: sumURL,
	35	EncodeRequestFunc: server.EncodeSumRequest,
	36	DecodeResponseFunc: server.DecodeSumResponse,
	37	}).Endpoint(),
	38	concat: (httptransport.Client{
	39	Client: c,
	40	Method: "GET",
	41	URL: concatURL,
	42	EncodeRequestFunc: server.EncodeConcatRequest,
	43	DecodeResponseFunc: server.DecodeConcatResponse,
	44	}).Endpoint(),
	45	}
	46	}
	47
	48	type client struct {
	49	context.Context
	50	log.Logger
	51	sum endpoint.Endpoint
	52	concat endpoint.Endpoint
	53	}
	54
	55	func (c client) Sum(a, b int) int {
	56	response, err := c.sum(c.Context, server.SumRequest{A: a, B: b})
	57	if err != nil {
	58	_ = c.Logger.Log("err", err)
	59	return 0
	60	}
	61	return response.(server.SumResponse).V
	62	}
	63
	64	func (c client) Concat(a, b string) string {
	65	response, err := c.concat(c.Context, server.ConcatRequest{A: a, B: b})
	66	if err != nil {
	67	_ = c.Logger.Log("err", err)
	68	return ""
	69	}
	70	return response.(server.ConcatResponse).V
	71	}

+145

-0

examples/addsvc/client/main.go less more

	0	package main
	1
	2	import (
	3	"flag"
	4	"fmt"
	5	"net/rpc"
	6	"net/url"
	7	"os"
	8	"path/filepath"
	9	"strconv"
	10	"strings"
	11	"time"
	12
	13	"github.com/apache/thrift/lib/go/thrift"
	14	"golang.org/x/net/context"
	15	"google.golang.org/grpc"
	16
	17	thriftadd "github.com/go-kit/kit/examples/addsvc/_thrift/gen-go/add"
	18	grpcclient "github.com/go-kit/kit/examples/addsvc/client/grpc"
	19	httpjsonclient "github.com/go-kit/kit/examples/addsvc/client/httpjson"
	20	netrpcclient "github.com/go-kit/kit/examples/addsvc/client/netrpc"
	21	thriftclient "github.com/go-kit/kit/examples/addsvc/client/thrift"
	22	"github.com/go-kit/kit/examples/addsvc/server"
	23	"github.com/go-kit/kit/log"
	24	)
	25
	26	func main() {
	27	var (
	28	transport = flag.String("transport", "httpjson", "httpjson, grpc, netrpc, thrift")
	29	httpAddr = flag.String("http.addr", "localhost:8001", "Address for HTTP (JSON) server")
	30	grpcAddr = flag.String("grpc.addr", "localhost:8002", "Address for gRPC server")
	31	netrpcAddr = flag.String("netrpc.addr", "localhost:8003", "Address for net/rpc server")
	32	thriftAddr = flag.String("thrift.addr", "localhost:8004", "Address for Thrift server")
	33	thriftProtocol = flag.String("thrift.protocol", "binary", "binary, compact, json, simplejson")
	34	thriftBufferSize = flag.Int("thrift.buffer.size", 0, "0 for unbuffered")
	35	thriftFramed = flag.Bool("thrift.framed", false, "true to enable framing")
	36	)
	37	flag.Parse()
	38	if len(os.Args) < 4 {
	39	fmt.Fprintf(os.Stderr, "\n%s [flags] method arg1 arg2\n\n", filepath.Base(os.Args[0]))
	40	flag.Usage()
	41	os.Exit(1)
	42	}
	43
	44	root := context.Background()
	45	method, s1, s2 := flag.Arg(0), flag.Arg(1), flag.Arg(2)
	46
	47	var logger log.Logger
	48	logger = log.NewLogfmtLogger(os.Stdout)
	49	logger = log.NewContext(logger).With("caller", log.DefaultCaller)
	50	logger = log.NewContext(logger).With("transport", *transport)
	51
	52	var svc server.AddService
	53	switch *transport {
	54	case "grpc":
	55	cc, err := grpc.Dial(*grpcAddr)
	56	if err != nil {
	57	_ = logger.Log("err", err)
	58	os.Exit(1)
	59	}
	60	defer cc.Close()
	61	svc = grpcclient.New(root, cc, logger)
	62
	63	case "httpjson":
	64	rawurl := *httpAddr
	65	if !strings.HasPrefix("http", rawurl) {
	66	rawurl = "http://" + rawurl
	67	}
	68	baseurl, err := url.Parse(rawurl)
	69	if err != nil {
	70	_ = logger.Log("err", err)
	71	os.Exit(1)
	72	}
	73	svc = httpjsonclient.New(root, baseurl, logger, nil)
	74
	75	case "netrpc":
	76	cli, err := rpc.DialHTTP("tcp", *netrpcAddr)
	77	if err != nil {
	78	_ = logger.Log("err", err)
	79	os.Exit(1)
	80	}
	81	defer cli.Close()
	82	svc = netrpcclient.New(cli, logger)
	83
	84	case "thrift":
	85	var protocolFactory thrift.TProtocolFactory
	86	switch *thriftProtocol {
	87	case "compact":
	88	protocolFactory = thrift.NewTCompactProtocolFactory()
	89	case "simplejson":
	90	protocolFactory = thrift.NewTSimpleJSONProtocolFactory()
	91	case "json":
	92	protocolFactory = thrift.NewTJSONProtocolFactory()
	93	case "binary", "":
	94	protocolFactory = thrift.NewTBinaryProtocolFactoryDefault()
	95	default:
	96	_ = logger.Log("protocol", *thriftProtocol, "err", "invalid protocol")
	97	os.Exit(1)
	98	}
	99	var transportFactory thrift.TTransportFactory
	100	if *thriftBufferSize > 0 {
	101	transportFactory = thrift.NewTBufferedTransportFactory(*thriftBufferSize)
	102	} else {
	103	transportFactory = thrift.NewTTransportFactory()
	104	}
	105	if *thriftFramed {
	106	transportFactory = thrift.NewTFramedTransportFactory(transportFactory)
	107	}
	108	transportSocket, err := thrift.NewTSocket(*thriftAddr)
	109	if err != nil {
	110	_ = logger.Log("during", "thrift.NewTSocket", "err", err)
	111	os.Exit(1)
	112	}
	113	trans := transportFactory.GetTransport(transportSocket)
	114	defer trans.Close()
	115	if err := trans.Open(); err != nil {
	116	_ = logger.Log("during", "thrift transport.Open", "err", err)
	117	os.Exit(1)
	118	}
	119	cli := thriftadd.NewAddServiceClientFactory(trans, protocolFactory)
	120	svc = thriftclient.New(cli, logger)
	121
	122	default:
	123	_ = logger.Log("err", "invalid transport")
	124	os.Exit(1)
	125	}
	126
	127	begin := time.Now()
	128	switch method {
	129	case "sum":
	130	a, _ := strconv.Atoi(s1)
	131	b, _ := strconv.Atoi(s2)
	132	v := svc.Sum(a, b)
	133	_ = logger.Log("method", "sum", "a", a, "b", b, "v", v, "took", time.Since(begin))
	134
	135	case "concat":
	136	a, b := s1, s2
	137	v := svc.Concat(a, b)
	138	_ = logger.Log("method", "concat", "a", a, "b", b, "v", v, "took", time.Since(begin))
	139
	140	default:
	141	_ = logger.Log("err", "invalid method "+method)
	142	os.Exit(1)
	143	}
	144	}

+29

-28

examples/addsvc/client/netrpc/client.go less more

2	2	import (
3	3	"net/rpc"
4	4
5		"golang.org/x/net/context"
6
7		"github.com/go-kit/kit/endpoint"
8		"github.com/go-kit/kit/examples/addsvc/reqrep"
	5	"github.com/go-kit/kit/examples/addsvc/server"
	6	"github.com/go-kit/kit/log"
9	7	)
10	8
11		// NewClient takes a net/rpc Client that should point to an instance of an
12		// addsvc. It returns an endpoint that wraps and invokes that Client.
13		func NewClient(c *rpc.Client) endpoint.Endpoint {
14		return func(ctx context.Context, request interface{}) (interface{}, error) {
15		var (
16		errs = make(chan error, 1)
17		responses = make(chan interface{}, 1)
18		)
19		go func() {
20		var response reqrep.AddResponse
21		if err := c.Call("addsvc.Add", request, &response); err != nil {
22		errs <- err
23		return
24		}
25		responses <- response
26		}()
27		select {
28		case <-ctx.Done():
29		return nil, context.DeadlineExceeded
30		case err := <-errs:
31		return nil, err
32		case response := <-responses:
33		return response, nil
34		}
	9	// New returns an AddService that's backed by the URL. baseurl will have its
	10	// scheme and hostport used, but its path will be overwritten. If client is
	11	// nil, http.DefaultClient will be used.
	12	func New(cli *rpc.Client, logger log.Logger) server.AddService {
	13	return client{cli, logger}
	14	}
	15
	16	type client struct {
	17	*rpc.Client
	18	log.Logger
	19	}
	20
	21	func (c client) Sum(a, b int) int {
	22	var reply server.SumResponse
	23	if err := c.Client.Call("addsvc.Sum", server.SumRequest{A: a, B: b}, &reply); err != nil {
	24	_ = c.Logger.Log("err", err)
	25	return 0
35	26	}
	27	return reply.V
36	28	}
	29
	30	func (c client) Concat(a, b string) string {
	31	var reply server.ConcatResponse
	32	if err := c.Client.Call("addsvc.Concat", server.ConcatRequest{A: a, B: b}, &reply); err != nil {
	33	_ = c.Logger.Log("err", err)
	34	return ""
	35	}
	36	return reply.V
	37	}

+27

-34

examples/addsvc/client/thrift/client.go less more

0	0	package thrift
1	1
2	2	import (
3		"golang.org/x/net/context"
4
5		"github.com/go-kit/kit/endpoint"
6	3	thriftadd "github.com/go-kit/kit/examples/addsvc/_thrift/gen-go/add"
7		"github.com/go-kit/kit/examples/addsvc/reqrep"
	4	"github.com/go-kit/kit/examples/addsvc/server"
	5	"github.com/go-kit/kit/log"
8	6	)
9	7
10		// NewClient takes a Thrift AddServiceClient, which should point to an
11		// instance of an addsvc. It returns an endpoint that wraps and invokes that
12		// client.
13		func NewClient(client *thriftadd.AddServiceClient) endpoint.Endpoint {
14		return func(ctx context.Context, request interface{}) (interface{}, error) {
15		var (
16		errs = make(chan error, 1)
17		responses = make(chan interface{}, 1)
18		)
19		go func() {
20		addReq, ok := request.(reqrep.AddRequest)
21		if !ok {
22		errs <- endpoint.ErrBadCast
23		return
24		}
25		reply, err := client.Add(addReq.A, addReq.B)
26		if err != nil {
27		errs <- err
28		return
29		}
30		responses <- reqrep.AddResponse{V: reply.Value}
31		}()
32		select {
33		case <-ctx.Done():
34		return nil, context.DeadlineExceeded
35		case err := <-errs:
36		return nil, err
37		case response := <-responses:
38		return response, nil
39		}
	8	// New returns an AddService that's backed by the Thrift client.
	9	func New(cli *thriftadd.AddServiceClient, logger log.Logger) server.AddService {
	10	return &client{cli, logger}
	11	}
	12
	13	type client struct {
	14	*thriftadd.AddServiceClient
	15	log.Logger
	16	}
	17
	18	func (c client) Sum(a, b int) int {
	19	reply, err := c.AddServiceClient.Sum(int64(a), int64(b))
	20	if err != nil {
	21	_ = c.Logger.Log("err", err)
	22	return 0
40	23	}
	24	return int(reply.Value)
41	25	}
	26
	27	func (c client) Concat(a, b string) string {
	28	reply, err := c.AddServiceClient.Concat(a, b)
	29	if err != nil {
	30	_ = c.Logger.Log("err", err)
	31	return ""
	32	}
	33	return reply.Value
	34	}

+13

-19

examples/addsvc/endpoint.go less more

3	3	"golang.org/x/net/context"
4	4
5	5	"github.com/go-kit/kit/endpoint"
6		"github.com/go-kit/kit/examples/addsvc/reqrep"
	6	"github.com/go-kit/kit/examples/addsvc/server"
7	7	)
8	8
9		// makeEndpoint returns an endpoint wrapping the passed Add. If Add were an
10		// interface with multiple methods, we'd need individual endpoints for each.
11		//
12		// This function is just boiler-plate; in theory, it could be generated.
13		func makeEndpoint(a Add) endpoint.Endpoint {
	9	func makeSumEndpoint(svc server.AddService) endpoint.Endpoint {
14	10	return func(ctx context.Context, request interface{}) (interface{}, error) {
15		select {
16		default:
17		case <-ctx.Done():
18		return nil, endpoint.ErrContextCanceled
19		}
20
21		addReq, ok := request.(reqrep.AddRequest)
22		if !ok {
23		return nil, endpoint.ErrBadCast
24		}
25
26		v := a(ctx, addReq.A, addReq.B)
27		return reqrep.AddResponse{V: v}, nil
	11	req := request.(server.SumRequest)
	12	v := svc.Sum(req.A, req.B)
	13	return server.SumResponse{V: v}, nil
28	14	}
29	15	}
	16
	17	func makeConcatEndpoint(svc server.AddService) endpoint.Endpoint {
	18	return func(ctx context.Context, request interface{}) (interface{}, error) {
	19	req := request.(server.ConcatRequest)
	20	v := svc.Concat(req.A, req.B)
	21	return server.ConcatResponse{V: v}, nil
	22	}
	23	}

-34

~~examples/addsvc/enhancements.go~~ less more

0		package main
1
2		import (
3		"time"
4
5		"golang.org/x/net/context"
6
7		"github.com/go-kit/kit/log"
8		"github.com/go-kit/kit/metrics"
9		)
10
11		func logging(logger log.Logger) func(Add) Add {
12		return func(next Add) Add {
13		return func(ctx context.Context, a, b int64) (v int64) {
14		defer func(begin time.Time) {
15		logger.Log("a", a, "b", b, "result", v, "took", time.Since(begin))
16		}(time.Now())
17		v = next(ctx, a, b)
18		return
19		}
20		}
21		}
22
23		func instrument(requests metrics.Counter, duration metrics.TimeHistogram) func(Add) Add {
24		return func(next Add) Add {
25		return func(ctx context.Context, a, b int64) int64 {
26		defer func(begin time.Time) {
27		requests.Add(1)
28		duration.Observe(time.Since(begin))
29		}(time.Now())
30		return next(ctx, a, b)
31		}
32		}
33		}

+10

-38

examples/addsvc/grpc_binding.go less more

2	2	import (
3	3	"golang.org/x/net/context"
4	4
5		"github.com/go-kit/kit/endpoint"
6	5	"github.com/go-kit/kit/examples/addsvc/pb"
7		"github.com/go-kit/kit/examples/addsvc/reqrep"
	6	"github.com/go-kit/kit/examples/addsvc/server"
8	7	)
9	8
10		// A binding wraps an Endpoint so that it's usable by a transport. grpcBinding
11		// makes an Endpoint usable over gRPC.
12		type grpcBinding struct{ endpoint.Endpoint }
	9	type grpcBinding struct {
	10	server.AddService
	11	}
13	12
14		// Add implements the proto3 AddServer by forwarding to the wrapped Endpoint.
15		//
16		// As far as I can tell, gRPC doesn't (currently) provide a user-accessible
17		// way to manipulate the RPC context, like headers for HTTP. So we don't have
18		// a way to transport e.g. Zipkin IDs with the request. TODO.
19		func (b grpcBinding) Add(ctx0 context.Context, req pb.AddRequest) (pb.AddReply, error) {
20		var (
21		ctx, cancel = context.WithCancel(ctx0)
22		errs = make(chan error, 1)
23		replies = make(chan *pb.AddReply, 1)
24		)
25		defer cancel()
26		go func() {
27		r, err := b.Endpoint(ctx, reqrep.AddRequest{A: req.A, B: req.B})
28		if err != nil {
29		errs <- err
30		return
31		}
32		resp, ok := r.(reqrep.AddResponse)
33		if !ok {
34		errs <- endpoint.ErrBadCast
35		return
36		}
37		replies <- &pb.AddReply{V: resp.V}
38		}()
39		select {
40		case <-ctx.Done():
41		return nil, context.DeadlineExceeded
42		case err := <-errs:
43		return nil, err
44		case reply := <-replies:
45		return reply, nil
46		}
	13	func (b grpcBinding) Sum(ctx context.Context, req pb.SumRequest) (pb.SumReply, error) {
	14	return &pb.SumReply{V: int64(b.AddService.Sum(int(req.A), int(req.B)))}, nil
47	15	}
	16
	17	func (b grpcBinding) Concat(ctx context.Context, req pb.ConcatRequest) (pb.ConcatReply, error) {
	18	return &pb.ConcatReply{V: b.AddService.Concat(req.A, req.B)}, nil
	19	}

-33

~~examples/addsvc/http_binding.go~~ less more

0		package main
1
2		import (
3		"encoding/json"
4		"net/http"
5
6		"golang.org/x/net/context"
7
8		"github.com/go-kit/kit/endpoint"
9		"github.com/go-kit/kit/examples/addsvc/reqrep"
10		httptransport "github.com/go-kit/kit/transport/http"
11		)
12
13		func makeHTTPBinding(ctx context.Context, e endpoint.Endpoint, before []httptransport.RequestFunc, after []httptransport.ResponseFunc) http.Handler {
14		decode := func(r *http.Request) (interface{}, error) {
15		var request reqrep.AddRequest
16		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
17		return nil, err
18		}
19		return request, nil
20		}
21		encode := func(w http.ResponseWriter, response interface{}) error {
22		return json.NewEncoder(w).Encode(response)
23		}
24		return httptransport.Server{
25		Context: ctx,
26		Endpoint: e,
27		DecodeRequestFunc: decode,
28		EncodeResponseFunc: encode,
29		Before: before,
30		After: append([]httptransport.ResponseFunc{httptransport.SetContentType("application/json; charset=utf-8")}, after...),
31		}
32		}

+128

-115

examples/addsvc/main.go less more

2	2	import (
3	3	"flag"
4	4	"fmt"
5		"io/ioutil"
6	5	stdlog "log"
7	6	"math/rand"
8	7	"net"
9	8	"net/http"
10		_ "net/http/pprof"
11	9	"net/rpc"
12	10	"os"
13	11	"os/signal"

22	20
23	21	"github.com/go-kit/kit/endpoint"
24	22	thriftadd "github.com/go-kit/kit/examples/addsvc/_thrift/gen-go/add"
25		httpclient "github.com/go-kit/kit/examples/addsvc/client/http"
26	23	"github.com/go-kit/kit/examples/addsvc/pb"
27		kitlog "github.com/go-kit/kit/log"
	24	"github.com/go-kit/kit/examples/addsvc/server"
	25	"github.com/go-kit/kit/log"
28	26	"github.com/go-kit/kit/metrics"
29	27	"github.com/go-kit/kit/metrics/expvar"
30	28	"github.com/go-kit/kit/metrics/prometheus"
31		"github.com/go-kit/kit/metrics/statsd"
32	29	"github.com/go-kit/kit/tracing/zipkin"
33	30	httptransport "github.com/go-kit/kit/transport/http"
34	31	)

38	35	// of glog. So, we define a new flag set, to keep those domains distinct.
39	36	fs := flag.NewFlagSet("", flag.ExitOnError)
40	37	var (
41		debugAddr = fs.String("debug.addr", ":8000", "Address for HTTP debug/instrumentation server")
42		httpAddr = fs.String("http.addr", ":8001", "Address for HTTP (JSON) server")
43		grpcAddr = fs.String("grpc.addr", ":8002", "Address for gRPC server")
44		netrpcAddr = fs.String("netrpc.addr", ":8003", "Address for net/rpc server")
45		thriftAddr = fs.String("thrift.addr", ":8004", "Address for Thrift server")
46		thriftProtocol = fs.String("thrift.protocol", "binary", "binary, compact, json, simplejson")
47		thriftBufferSize = fs.Int("thrift.buffer.size", 0, "0 for unbuffered")
48		thriftFramed = fs.Bool("thrift.framed", false, "true to enable framing")
49
50		proxyHTTPURL = fs.String("proxy.http.url", "", "if set, proxy requests over HTTP to this addsvc")
51
	38	debugAddr = fs.String("debug.addr", ":8000", "Address for HTTP debug/instrumentation server")
	39	httpAddr = fs.String("http.addr", ":8001", "Address for HTTP (JSON) server")
	40	grpcAddr = fs.String("grpc.addr", ":8002", "Address for gRPC server")
	41	netrpcAddr = fs.String("netrpc.addr", ":8003", "Address for net/rpc server")
	42	thriftAddr = fs.String("thrift.addr", ":8004", "Address for Thrift server")
	43	thriftProtocol = fs.String("thrift.protocol", "binary", "binary, compact, json, simplejson")
	44	thriftBufferSize = fs.Int("thrift.buffer.size", 0, "0 for unbuffered")
	45	thriftFramed = fs.Bool("thrift.framed", false, "true to enable framing")
	46	zipkinHostPort = fs.String("zipkin.host.port", "my.service.domain:12345", "Zipkin host:port")
52	47	zipkinServiceName = fs.String("zipkin.service.name", "addsvc", "Zipkin service name")
53	48	zipkinCollectorAddr = fs.String("zipkin.collector.addr", "", "Zipkin Scribe collector address (empty will log spans)")
54	49	zipkinCollectorTimeout = fs.Duration("zipkin.collector.timeout", time.Second, "Zipkin collector timeout")

56	51	zipkinCollectorBatchInterval = fs.Duration("zipkin.collector.batch.interval", time.Second, "Zipkin collector batch interval")
57	52	)
58	53	flag.Usage = fs.Usage // only show our flags
59		fs.Parse(os.Args[1:])
60
61		// `package log` domain
62		var logger kitlog.Logger
63		logger = kitlog.NewLogfmtLogger(os.Stderr)
64		logger = kitlog.NewContext(logger).With("ts", kitlog.DefaultTimestampUTC)
65		stdlog.SetOutput(kitlog.NewStdlibAdapter(logger)) // redirect stdlib logging to us
66		stdlog.SetFlags(0) // flags are handled in our logger
67
68		// `package metrics` domain
69		requests := metrics.NewMultiCounter(
70		expvar.NewCounter("requests"),
71		statsd.NewCounter(ioutil.Discard, "requests_total", time.Second),
72		prometheus.NewCounter(stdprometheus.CounterOpts{
73		Namespace: "addsvc",
74		Subsystem: "add",
75		Name: "requests_total",
76		Help: "Total number of received requests.",
77		}, []string{}),
78		)
79		duration := metrics.NewTimeHistogram(time.Nanosecond, metrics.NewMultiHistogram(
80		expvar.NewHistogram("duration_nanoseconds_total", 0, 1e9, 3, 50, 95, 99),
81		statsd.NewHistogram(ioutil.Discard, "duration_nanoseconds_total", time.Second),
82		prometheus.NewSummary(stdprometheus.SummaryOpts{
83		Namespace: "addsvc",
84		Subsystem: "add",
85		Name: "duration_nanoseconds_total",
86		Help: "Total nanoseconds spend serving requests.",
87		}, []string{}),
88		))
89
90		// `package tracing` domain
91		zipkinHostPort := "localhost:1234" // TODO Zipkin makes overly simple assumptions about services
92		var zipkinCollector zipkin.Collector = loggingCollector{logger}
93		if *zipkinCollectorAddr != "" {
94		var err error
95		if zipkinCollector, err = zipkin.NewScribeCollector(
96		*zipkinCollectorAddr,
97		*zipkinCollectorTimeout,
98		zipkin.ScribeBatchSize(*zipkinCollectorBatchSize),
99		zipkin.ScribeBatchInterval(*zipkinCollectorBatchInterval),
100		zipkin.ScribeLogger(logger),
101		); err != nil {
102		logger.Log("err", err)
103		os.Exit(1)
104		}
105		}
106		zipkinMethodName := "add"
107		zipkinSpanFunc := zipkin.MakeNewSpanFunc(zipkinHostPort, *zipkinServiceName, zipkinMethodName)
108
109		// Our business and operational domain
110		var a Add = pureAdd
111		if *proxyHTTPURL != "" {
112		var e endpoint.Endpoint
113		e = httpclient.NewClient("GET", *proxyHTTPURL, zipkin.ToRequest(zipkinSpanFunc))
114		e = zipkin.AnnotateClient(zipkinSpanFunc, zipkinCollector)(e)
115		a = proxyAdd(e, logger)
116		}
117		a = logging(logger)(a)
118		a = instrument(requests, duration)(a)
119
120		// Server domain
121		var e endpoint.Endpoint
122		e = makeEndpoint(a)
123		e = zipkin.AnnotateServer(zipkinSpanFunc, zipkinCollector)(e)
	54	if err := fs.Parse(os.Args[1:]); err != nil {
	55	fmt.Fprintf(os.Stderr, "%v", err)
	56	os.Exit(1)
	57	}
	58
	59	// package log
	60	var logger log.Logger
	61	{
	62	logger = log.NewLogfmtLogger(os.Stderr)
	63	logger = log.NewContext(logger).With("ts", log.DefaultTimestampUTC).With("caller", log.DefaultCaller)
	64	stdlog.SetFlags(0) // flags are handled by Go kit's logger
	65	stdlog.SetOutput(log.NewStdlibAdapter(logger)) // redirect anything using stdlib log to us
	66	}
	67
	68	// package metrics
	69	var requestDuration metrics.TimeHistogram
	70	{
	71	requestDuration = metrics.NewTimeHistogram(time.Nanosecond, metrics.NewMultiHistogram(
	72	expvar.NewHistogram("request_duration_ns", 0, 5e9, 1, 50, 95, 99),
	73	prometheus.NewSummary(stdprometheus.SummaryOpts{
	74	Namespace: "myorg",
	75	Subsystem: "addsvc",
	76	Name: "duration_ns",
	77	Help: "Request duration in nanoseconds.",
	78	}, []string{"method"}),
	79	))
	80	}
	81
	82	// package tracing
	83	var collector zipkin.Collector
	84	{
	85	zipkinLogger := log.NewContext(logger).With("component", "zipkin")
	86	collector = loggingCollector{zipkinLogger} // TODO(pb)
	87	if *zipkinCollectorAddr != "" {
	88	var err error
	89	if collector, err = zipkin.NewScribeCollector(
	90	*zipkinCollectorAddr,
	91	*zipkinCollectorTimeout,
	92	zipkin.ScribeBatchSize(*zipkinCollectorBatchSize),
	93	zipkin.ScribeBatchInterval(*zipkinCollectorBatchInterval),
	94	zipkin.ScribeLogger(zipkinLogger),
	95	); err != nil {
	96	_ = zipkinLogger.Log("err", err)
	97	os.Exit(1)
	98	}
	99	}
	100	}
	101
	102	// Business domain
	103	var svc server.AddService
	104	{
	105	svc = pureAddService{}
	106	svc = loggingMiddleware{svc, logger}
	107	svc = instrumentingMiddleware{svc, requestDuration}
	108	}
124	109
125	110	// Mechanical stuff
126	111	rand.Seed(time.Now().UnixNano())

131	116	errc <- interrupt()
132	117	}()
133	118
134		// Transport: HTTP (debug/instrumentation)
135		go func() {
136		logger.Log("addr", *debugAddr, "transport", "debug")
137		errc <- http.ListenAndServe(*debugAddr, nil)
138		}()
139
140		// Transport: HTTP (JSON)
141		go func() {
142		ctx, cancel := context.WithCancel(root)
143		defer cancel()
144		before := []httptransport.RequestFunc{zipkin.ToContext(zipkinSpanFunc, logger)}
145		after := []httptransport.ResponseFunc{}
146		handler := makeHTTPBinding(ctx, e, before, after)
147		logger.Log("addr", *httpAddr, "transport", "HTTP/JSON")
148		errc <- http.ListenAndServe(*httpAddr, handler)
	119	// Debug/instrumentation
	120	go func() {
	121	transportLogger := log.NewContext(logger).With("transport", "debug")
	122	_ = transportLogger.Log("addr", *debugAddr)
	123	errc <- http.ListenAndServe(*debugAddr, nil) // DefaultServeMux
	124	}()
	125
	126	// Transport: HTTP/JSON
	127	go func() {
	128	var (
	129	transportLogger = log.NewContext(logger).With("transport", "HTTP/JSON")
	130	tracingLogger = log.NewContext(transportLogger).With("component", "tracing")
	131	newSumSpan = zipkin.MakeNewSpanFunc(zipkinHostPort, zipkinServiceName, "sum")
	132	newConcatSpan = zipkin.MakeNewSpanFunc(zipkinHostPort, zipkinServiceName, "concat")
	133	traceSum = zipkin.ToContext(newSumSpan, tracingLogger)
	134	traceConcat = zipkin.ToContext(newConcatSpan, tracingLogger)
	135	mux = http.NewServeMux()
	136	sum, concat endpoint.Endpoint
	137	)
	138
	139	sum = makeSumEndpoint(svc)
	140	sum = zipkin.AnnotateServer(newSumSpan, collector)(sum)
	141	mux.Handle("/sum", httptransport.Server{
	142	Context: root,
	143	Endpoint: sum,
	144	DecodeRequestFunc: server.DecodeSumRequest,
	145	EncodeResponseFunc: server.EncodeSumResponse,
	146	Before: []httptransport.RequestFunc{traceSum},
	147	After: []httptransport.ResponseFunc{},
	148	Logger: transportLogger,
	149	})
	150
	151	concat = makeConcatEndpoint(svc)
	152	concat = zipkin.AnnotateServer(newConcatSpan, collector)(concat)
	153	mux.Handle("/concat", httptransport.Server{
	154	Context: root,
	155	Endpoint: concat,
	156	DecodeRequestFunc: server.DecodeConcatRequest,
	157	EncodeResponseFunc: server.EncodeConcatResponse,
	158	Before: []httptransport.RequestFunc{traceConcat},
	159	After: []httptransport.ResponseFunc{},
	160	Logger: transportLogger,
	161	})
	162
	163	_ = transportLogger.Log("addr", *httpAddr)
	164	errc <- http.ListenAndServe(*httpAddr, mux)
149	165	}()
150	166
151	167	// Transport: gRPC
152	168	go func() {
	169	transportLogger := log.NewContext(logger).With("transport", "gRPC")
153	170	ln, err := net.Listen("tcp", *grpcAddr)
154	171	if err != nil {
155	172	errc <- err
156	173	return
157	174	}
158		s := grpc.NewServer() // uses its own context?
159		pb.RegisterAddServer(s, grpcBinding{e})
160		logger.Log("addr", *grpcAddr, "transport", "gRPC")
	175	s := grpc.NewServer() // uses its own, internal context
	176	pb.RegisterAddServer(s, grpcBinding{svc})
	177	_ = transportLogger.Log("addr", *grpcAddr)
161	178	errc <- s.Serve(ln)
162	179	}()
163	180
164	181	// Transport: net/rpc
165	182	go func() {
166		ctx, cancel := context.WithCancel(root)
167		defer cancel()
	183	transportLogger := log.NewContext(logger).With("transport", "net/rpc")
168	184	s := rpc.NewServer()
169		s.RegisterName("addsvc", NetrpcBinding{ctx, e})
	185	if err := s.RegisterName("addsvc", netrpcBinding{svc}); err != nil {
	186	errc <- err
	187	return
	188	}
170	189	s.HandleHTTP(rpc.DefaultRPCPath, rpc.DefaultDebugPath)
171		logger.Log("addr", *netrpcAddr, "transport", "net/rpc")
	190	_ = transportLogger.Log("addr", *netrpcAddr)
172	191	errc <- http.ListenAndServe(*netrpcAddr, s)
173	192	}()
174	193
175	194	// Transport: Thrift
176	195	go func() {
177		ctx, cancel := context.WithCancel(root)
178		defer cancel()
179
180	196	var protocolFactory thrift.TProtocolFactory
181	197	switch *thriftProtocol {
182	198	case "binary":

191	207	errc <- fmt.Errorf("invalid Thrift protocol %q", *thriftProtocol)
192	208	return
193	209	}
194
195	210	var transportFactory thrift.TTransportFactory
196	211	if *thriftBufferSize > 0 {
197	212	transportFactory = thrift.NewTBufferedTransportFactory(*thriftBufferSize)
198	213	} else {
199	214	transportFactory = thrift.NewTTransportFactory()
200	215	}
201
202	216	if *thriftFramed {
203	217	transportFactory = thrift.NewTFramedTransportFactory(transportFactory)
204	218	}
205
206	219	transport, err := thrift.NewTServerSocket(*thriftAddr)
207	220	if err != nil {
208	221	errc <- err
209	222	return
210	223	}
211
212		logger.Log("addr", *thriftAddr, "transport", "Thrift")
	224	transportLogger := log.NewContext(logger).With("transport", "net/rpc")
	225	_ = transportLogger.Log("addr", *thriftAddr)
213	226	errc <- thrift.NewTSimpleServer4(
214		thriftadd.NewAddServiceProcessor(thriftBinding{ctx, e}),
	227	thriftadd.NewAddServiceProcessor(thriftBinding{svc}),
215	228	transport,
216	229	transportFactory,
217	230	protocolFactory,
218	231	).Serve()
219	232	}()
220	233
221		logger.Log("fatal", <-errc)
	234	_ = logger.Log("fatal", <-errc)
222	235	}
223	236
224	237	func interrupt() error {

227	240	return fmt.Errorf("%s", <-c)
228	241	}
229	242
230		type loggingCollector struct{ kitlog.Logger }
	243	type loggingCollector struct{ log.Logger }
231	244
232	245	func (c loggingCollector) Collect(s *zipkin.Span) error {
233	246	annotations := s.Encode().GetAnnotations()

235	248	for i, a := range annotations {
236	249	values[i] = a.Value
237	250	}
238		c.Logger.Log(
	251	_ = c.Logger.Log(
239	252	"trace_id", s.TraceID(),
240	253	"span_id", s.SpanID(),
241	254	"parent_span_id", s.ParentSpanID(),

+13

-39

examples/addsvc/netrpc_binding.go less more

0	0	package main
1	1
2	2	import (
3		"golang.org/x/net/context"
4
5		"github.com/go-kit/kit/endpoint"
6		"github.com/go-kit/kit/examples/addsvc/reqrep"
	3	"github.com/go-kit/kit/examples/addsvc/server"
7	4	)
8	5
9		// NetrpcBinding makes an endpoint usable over net/rpc. It needs to be
10		// exported to be picked up by net/rpc.
11		type NetrpcBinding struct {
12		ctx context.Context // has methods which should not be made available
13		endpoint.Endpoint
	6	type netrpcBinding struct {
	7	server.AddService
14	8	}
15	9
16		// Add implements the net/rpc method definition.
17		func (b NetrpcBinding) Add(request reqrep.AddRequest, response *reqrep.AddResponse) error {
18		var (
19		ctx, cancel = context.WithCancel(b.ctx)
20		errs = make(chan error, 1)
21		responses = make(chan reqrep.AddResponse, 1)
22		)
23		defer cancel()
24		go func() {
25		resp, err := b.Endpoint(ctx, request)
26		if err != nil {
27		errs <- err
28		return
29		}
30		addResp, ok := resp.(reqrep.AddResponse)
31		if !ok {
32		errs <- endpoint.ErrBadCast
33		return
34		}
35		responses <- addResp
36		}()
37		select {
38		case <-ctx.Done():
39		return context.DeadlineExceeded
40		case err := <-errs:
41		return err
42		case resp := <-responses:
43		(*response) = resp
44		return nil
45		}
	10	func (b netrpcBinding) Sum(request server.SumRequest, response *server.SumResponse) error {
	11	v := b.AddService.Sum(request.A, request.B)
	12	(*response) = server.SumResponse{V: v}
	13	return nil
46	14	}
	15
	16	func (b netrpcBinding) Concat(request server.ConcatRequest, response *server.ConcatResponse) error {
	17	v := b.AddService.Concat(request.A, request.B)
	18	(*response) = server.ConcatResponse{V: v}
	19	return nil
	20	}

+81

-30

examples/addsvc/pb/add.pb.go less more

8	8	add.proto
9	9
10	10	It has these top-level messages:
11		AddRequest
12		AddReply
	11	SumRequest
	12	SumReply
	13	ConcatRequest
	14	ConcatReply
13	15	*/
14	16	package pb
15	17
16	18	import proto "github.com/golang/protobuf/proto"
	19	import fmt "fmt"
	20	import math "math"
17	21
18	22	import (
19	23	context "golang.org/x/net/context"

21	25	)
22	26
23	27	// Reference imports to suppress errors if they are not otherwise used.
24		var _ context.Context
25		var _ grpc.ClientConn
	28	var _ = proto.Marshal
	29	var _ = fmt.Errorf
	30	var _ = math.Inf
26	31
27		// Reference imports to suppress errors if they are not otherwise used.
28		var _ = proto.Marshal
29
30		// The request contains two parameters.
31		type AddRequest struct {
	32	// The sum request contains two parameters.
	33	type SumRequest struct {
32	34	A int64 `protobuf:"varint,1,opt,name=a" json:"a,omitempty"`
33	35	B int64 `protobuf:"varint,2,opt,name=b" json:"b,omitempty"`
34	36	}
35	37
36		func (m AddRequest) Reset() { m = AddRequest{} }
37		func (m *AddRequest) String() string { return proto.CompactTextString(m) }
38		func (*AddRequest) ProtoMessage() {}
	38	func (m SumRequest) Reset() { m = SumRequest{} }
	39	func (m *SumRequest) String() string { return proto.CompactTextString(m) }
	40	func (*SumRequest) ProtoMessage() {}
39	41
40		// The response contains the result of the calculation.
41		type AddReply struct {
	42	// The sum response contains the result of the calculation.
	43	type SumReply struct {
42	44	V int64 `protobuf:"varint,1,opt,name=v" json:"v,omitempty"`
43	45	}
44	46
45		func (m AddReply) Reset() { m = AddReply{} }
46		func (m *AddReply) String() string { return proto.CompactTextString(m) }
47		func (*AddReply) ProtoMessage() {}
	47	func (m SumReply) Reset() { m = SumReply{} }
	48	func (m *SumReply) String() string { return proto.CompactTextString(m) }
	49	func (*SumReply) ProtoMessage() {}
48	50
49		func init() {
	51	// The Concat request contains two parameters.
	52	type ConcatRequest struct {
	53	A string `protobuf:"bytes,1,opt,name=a" json:"a,omitempty"`
	54	B string `protobuf:"bytes,2,opt,name=b" json:"b,omitempty"`
50	55	}
	56
	57	func (m ConcatRequest) Reset() { m = ConcatRequest{} }
	58	func (m *ConcatRequest) String() string { return proto.CompactTextString(m) }
	59	func (*ConcatRequest) ProtoMessage() {}
	60
	61	// The Concat response contains the result of the concatenation.
	62	type ConcatReply struct {
	63	V string `protobuf:"bytes,1,opt,name=v" json:"v,omitempty"`
	64	}
	65
	66	func (m ConcatReply) Reset() { m = ConcatReply{} }
	67	func (m *ConcatReply) String() string { return proto.CompactTextString(m) }
	68	func (*ConcatReply) ProtoMessage() {}
	69
	70	// Reference imports to suppress errors if they are not otherwise used.
	71	var _ context.Context
	72	var _ grpc.ClientConn
51	73
52	74	// Client API for Add service
53	75
54	76	type AddClient interface {
55		// Adds two integers.
56		Add(ctx context.Context, in AddRequest, opts ...grpc.CallOption) (AddReply, error)
	77	// Sums two integers.
	78	Sum(ctx context.Context, in SumRequest, opts ...grpc.CallOption) (SumReply, error)
	79	// Concatenates two strings
	80	Concat(ctx context.Context, in ConcatRequest, opts ...grpc.CallOption) (ConcatReply, error)
57	81	}
58	82
59	83	type addClient struct {

64	88	return &addClient{cc}
65	89	}
66	90
67		func (c addClient) Add(ctx context.Context, in AddRequest, opts ...grpc.CallOption) (*AddReply, error) {
68		out := new(AddReply)
69		err := grpc.Invoke(ctx, "/pb.Add/Add", in, out, c.cc, opts...)
	91	func (c addClient) Sum(ctx context.Context, in SumRequest, opts ...grpc.CallOption) (*SumReply, error) {
	92	out := new(SumReply)
	93	err := grpc.Invoke(ctx, "/pb.Add/Sum", in, out, c.cc, opts...)
	94	if err != nil {
	95	return nil, err
	96	}
	97	return out, nil
	98	}
	99
	100	func (c addClient) Concat(ctx context.Context, in ConcatRequest, opts ...grpc.CallOption) (*ConcatReply, error) {
	101	out := new(ConcatReply)
	102	err := grpc.Invoke(ctx, "/pb.Add/Concat", in, out, c.cc, opts...)
70	103	if err != nil {
71	104	return nil, err
72	105	}

76	109	// Server API for Add service
77	110
78	111	type AddServer interface {
79		// Adds two integers.
80		Add(context.Context, AddRequest) (AddReply, error)
	112	// Sums two integers.
	113	Sum(context.Context, SumRequest) (SumReply, error)
	114	// Concatenates two strings
	115	Concat(context.Context, ConcatRequest) (ConcatReply, error)
81	116	}
82	117
83	118	func RegisterAddServer(s *grpc.Server, srv AddServer) {
84	119	s.RegisterService(&_Add_serviceDesc, srv)
85	120	}
86	121
87		func _Add_Add_Handler(srv interface{}, ctx context.Context, codec grpc.Codec, buf []byte) (interface{}, error) {
88		in := new(AddRequest)
	122	func _Add_Sum_Handler(srv interface{}, ctx context.Context, codec grpc.Codec, buf []byte) (interface{}, error) {
	123	in := new(SumRequest)
89	124	if err := codec.Unmarshal(buf, in); err != nil {
90	125	return nil, err
91	126	}
92		out, err := srv.(AddServer).Add(ctx, in)
	127	out, err := srv.(AddServer).Sum(ctx, in)
	128	if err != nil {
	129	return nil, err
	130	}
	131	return out, nil
	132	}
	133
	134	func _Add_Concat_Handler(srv interface{}, ctx context.Context, codec grpc.Codec, buf []byte) (interface{}, error) {
	135	in := new(ConcatRequest)
	136	if err := codec.Unmarshal(buf, in); err != nil {
	137	return nil, err
	138	}
	139	out, err := srv.(AddServer).Concat(ctx, in)
93	140	if err != nil {
94	141	return nil, err
95	142	}

101	148	HandlerType: (*AddServer)(nil),
102	149	Methods: []grpc.MethodDesc{
103	150	{
104		MethodName: "Add",
105		Handler: _Add_Add_Handler,
	151	MethodName: "Sum",
	152	Handler: _Add_Sum_Handler,
	153	},
	154	{
	155	MethodName: "Concat",
	156	Handler: _Add_Concat_Handler,
106	157	},
107	158	},
108	159	Streams: []grpc.StreamDesc{},

+20

-6

examples/addsvc/pb/add.proto less more

3	3
4	4	// The Add service definition.
5	5	service Add {
6		// Adds two integers.
7		rpc Add (AddRequest) returns (AddReply) {}
	6	// Sums two integers.
	7	rpc Sum (SumRequest) returns (SumReply) {}
	8
	9	// Concatenates two strings
	10	rpc Concat (ConcatRequest) returns (ConcatReply) {}
8	11	}
9	12
10		// The request contains two parameters.
11		message AddRequest {
	13	// The sum request contains two parameters.
	14	message SumRequest {
12	15	int64 a = 1;
13	16	int64 b = 2;
14	17	}
15	18
16		// The response contains the result of the calculation.
17		message AddReply {
	19	// The sum response contains the result of the calculation.
	20	message SumReply {
18	21	int64 v = 1;
19	22	}
	23
	24	// The Concat request contains two parameters.
	25	message ConcatRequest {
	26	string a = 1;
	27	string b = 2;
	28	}
	29
	30	// The Concat response contains the result of the concatenation.
	31	message ConcatReply {
	32	string v = 1;
	33	}

-2

examples/addsvc/pb/compile.sh less more

0	0	#!/usr/bin/env sh
1	1
2		# Update protoc via
	2	# Install proto3 from source
	3	# brew install autoconf automake libtool
	4	# git clone https://github.com/google/protobuf
	5	# ./autogen.sh ; ./configure ; make ; make install
	6	#
	7	# Update protoc Go bindings via
3	8	# go get -u github.com/golang/protobuf/{proto,protoc-gen-go}
4	9	#
5	10	# See also
6		# https://github.com/grpc/grpc-common/tree/master/go
	11	# https://github.com/grpc/grpc-go/tree/master/examples
7	12
8	13	protoc add.proto --go_out=plugins=grpc:.

-16

~~examples/addsvc/reqrep/request_response.go~~ less more

0		package reqrep
1
2		// The concrete request and response types are defined for each method our
3		// service implements. Request types should be annotated sufficiently for all
4		// transports we intend to use.
5
6		// AddRequest is a request for the add method.
7		type AddRequest struct {
8		A int64 `json:"a"`
9		B int64 `json:"b"`
10		}
11
12		// AddResponse is a response to the add method.
13		type AddResponse struct {
14		V int64 `json:"v"`
15		}

+82

-0

examples/addsvc/server/encode_decode.go less more

	0	package server
	1
	2	import (
	3	"bytes"
	4	"encoding/json"
	5	"io/ioutil"
	6	"net/http"
	7	)
	8
	9	// DecodeSumRequest decodes the request from the provided HTTP request, simply
	10	// by JSON decoding from the request body. It's designed to be used in
	11	// transport/http.Server.
	12	func DecodeSumRequest(r *http.Request) (interface{}, error) {
	13	var request SumRequest
	14	err := json.NewDecoder(r.Body).Decode(&request)
	15	return request, err
	16	}
	17
	18	// EncodeSumResponse encodes the response to the provided HTTP response
	19	// writer, simply by JSON encoding to the writer. It's designed to be used in
	20	// transport/http.Server.
	21	func EncodeSumResponse(w http.ResponseWriter, response interface{}) error {
	22	return json.NewEncoder(w).Encode(response)
	23	}
	24
	25	// DecodeConcatRequest decodes the request from the provided HTTP request,
	26	// simply by JSON decoding from the request body. It's designed to be used in
	27	// transport/http.Server.
	28	func DecodeConcatRequest(r *http.Request) (interface{}, error) {
	29	var request ConcatRequest
	30	err := json.NewDecoder(r.Body).Decode(&request)
	31	return request, err
	32	}
	33
	34	// EncodeConcatResponse encodes the response to the provided HTTP response
	35	// writer, simply by JSON encoding to the writer. It's designed to be used in
	36	// transport/http.Server.
	37	func EncodeConcatResponse(w http.ResponseWriter, response interface{}) error {
	38	return json.NewEncoder(w).Encode(response)
	39	}
	40
	41	// EncodeSumRequest encodes the request to the provided HTTP request, simply
	42	// by JSON encoding to the request body. It's designed to be used in
	43	// transport/http.Client.
	44	func EncodeSumRequest(r *http.Request, request interface{}) error {
	45	var buf bytes.Buffer
	46	if err := json.NewEncoder(&buf).Encode(request); err != nil {
	47	return err
	48	}
	49	r.Body = ioutil.NopCloser(&buf)
	50	return nil
	51	}
	52
	53	// DecodeSumResponse decodes the response from the provided HTTP response,
	54	// simply by JSON decoding from the response body. It's designed to be used in
	55	// transport/http.Client.
	56	func DecodeSumResponse(resp *http.Response) (interface{}, error) {
	57	var response SumResponse
	58	err := json.NewDecoder(resp.Body).Decode(&response)
	59	return response, err
	60	}
	61
	62	// EncodeConcatRequest encodes the request to the provided HTTP request,
	63	// simply by JSON encoding to the request body. It's designed to be used in
	64	// transport/http.Client.
	65	func EncodeConcatRequest(r *http.Request, request interface{}) error {
	66	var buf bytes.Buffer
	67	if err := json.NewEncoder(&buf).Encode(request); err != nil {
	68	return err
	69	}
	70	r.Body = ioutil.NopCloser(&buf)
	71	return nil
	72	}
	73
	74	// DecodeConcatResponse decodes the response from the provided HTTP response,
	75	// simply by JSON decoding from the response body. It's designed to be used in
	76	// transport/http.Client.
	77	func DecodeConcatResponse(resp *http.Response) (interface{}, error) {
	78	var response ConcatResponse
	79	err := json.NewDecoder(resp.Body).Decode(&response)
	80	return response, err
	81	}

+23

-0

examples/addsvc/server/request_response.go less more

	0	package server
	1
	2	// SumRequest is the business domain type for a Sum method request.
	3	type SumRequest struct {
	4	A int `json:"a"`
	5	B int `json:"b"`
	6	}
	7
	8	// SumResponse is the business domain type for a Sum method response.
	9	type SumResponse struct {
	10	V int `json:"v"`
	11	}
	12
	13	// ConcatRequest is the business domain type for a Concat method request.
	14	type ConcatRequest struct {
	15	A string `json:"a"`
	16	B string `json:"b"`
	17	}
	18
	19	// ConcatResponse is the business domain type for a Concat method response.
	20	type ConcatResponse struct {
	21	V string `json:"v"`
	22	}

-0

examples/addsvc/server/service.go less more

	0	package server
	1
	2	// AddService is the abstract representation of this service.
	3	type AddService interface {
	4	Sum(a, b int) int
	5	Concat(a, b string) string
	6	}

+71

-0

examples/addsvc/service.go less more

	0	package main
	1
	2	import (
	3	"time"
	4
	5	"github.com/go-kit/kit/examples/addsvc/server"
	6	"github.com/go-kit/kit/log"
	7	"github.com/go-kit/kit/metrics"
	8	)
	9
	10	type pureAddService struct{}
	11
	12	func (pureAddService) Sum(a, b int) int { return a + b }
	13
	14	func (pureAddService) Concat(a, b string) string { return a + b }
	15
	16	type loggingMiddleware struct {
	17	server.AddService
	18	log.Logger
	19	}
	20
	21	func (m loggingMiddleware) Sum(a, b int) (v int) {
	22	defer func(begin time.Time) {
	23	_ = m.Logger.Log(
	24	"method", "sum",
	25	"a", a,
	26	"b", b,
	27	"v", v,
	28	"took", time.Since(begin),
	29	)
	30	}(time.Now())
	31	v = m.AddService.Sum(a, b)
	32	return
	33	}
	34
	35	func (m loggingMiddleware) Concat(a, b string) (v string) {
	36	defer func(begin time.Time) {
	37	_ = m.Logger.Log(
	38	"method", "concat",
	39	"a", a,
	40	"b", b,
	41	"v", v,
	42	"took", time.Since(begin),
	43	)
	44	}(time.Now())
	45	v = m.AddService.Concat(a, b)
	46	return
	47	}
	48
	49	type instrumentingMiddleware struct {
	50	server.AddService
	51	requestDuration metrics.TimeHistogram
	52	}
	53
	54	func (m instrumentingMiddleware) Sum(a, b int) (v int) {
	55	defer func(begin time.Time) {
	56	methodField := metrics.Field{Key: "method", Value: "sum"}
	57	m.requestDuration.With(methodField).Observe(time.Since(begin))
	58	}(time.Now())
	59	v = m.AddService.Sum(a, b)
	60	return
	61	}
	62
	63	func (m instrumentingMiddleware) Concat(a, b string) (v string) {
	64	defer func(begin time.Time) {
	65	methodField := metrics.Field{Key: "method", Value: "concat"}
	66	m.requestDuration.With(methodField).Observe(time.Since(begin))
	67	}(time.Now())
	68	v = m.AddService.Concat(a, b)
	69	return
	70	}

+10

-37

examples/addsvc/thrift_binding.go less more

0	0	package main
1	1
2	2	import (
3		"golang.org/x/net/context"
4
5		"github.com/go-kit/kit/endpoint"
6	3	thriftadd "github.com/go-kit/kit/examples/addsvc/_thrift/gen-go/add"
7		"github.com/go-kit/kit/examples/addsvc/reqrep"
	4	"github.com/go-kit/kit/examples/addsvc/server"
8	5	)
9	6
10		// A binding wraps an Endpoint so that it's usable by a transport.
11		// thriftBinding makes an Endpoint usable over Thrift.
12	7	type thriftBinding struct {
13		context.Context
14		endpoint.Endpoint
	8	server.AddService
15	9	}
16	10
17		// Add implements Thrift's AddService interface.
18		func (tb thriftBinding) Add(a, b int64) (*thriftadd.AddReply, error) {
19		var (
20		ctx, cancel = context.WithCancel(tb.Context)
21		errs = make(chan error, 1)
22		replies = make(chan *thriftadd.AddReply, 1)
23		)
24		defer cancel()
25		go func() {
26		r, err := tb.Endpoint(ctx, reqrep.AddRequest{A: a, B: b})
27		if err != nil {
28		errs <- err
29		return
30		}
31		resp, ok := r.(reqrep.AddResponse)
32		if !ok {
33		errs <- endpoint.ErrBadCast
34		return
35		}
36		replies <- &thriftadd.AddReply{Value: resp.V}
37		}()
38		select {
39		case <-ctx.Done():
40		return nil, context.DeadlineExceeded
41		case err := <-errs:
42		return nil, err
43		case reply := <-replies:
44		return reply, nil
45		}
	11	func (tb thriftBinding) Sum(a, b int64) (*thriftadd.SumReply, error) {
	12	v := tb.AddService.Sum(int(a), int(b))
	13	return &thriftadd.SumReply{Value: int64(v)}, nil
46	14	}
	15
	16	func (tb thriftBinding) Concat(a, b string) (*thriftadd.ConcatReply, error) {
	17	v := tb.AddService.Concat(a, b)
	18	return &thriftadd.ConcatReply{Value: v}, nil
	19	}

-1

examples/stringsvc3/proxying.go less more

84	84	Client: http.DefaultClient,
85	85	Method: "GET",
86	86	URL: u,
87		Context: ctx,
88	87	DecodeResponseFunc: decodeUppercaseResponse,
89	88	EncodeRequestFunc: encodeRequest,
90	89	}).Endpoint()

~~examples/stringsvc3/stringsvc3~~ less more

Binary diff not shown

-3

transport/http/client.go less more

19	19
20	20	// URL must be provided.
21	21	URL *url.URL
22
23		// A background context must be provided.
24		context.Context
25	22
26	23	// EncodeRequestFunc must be provided. The HTTP request passed to the
27	24	// EncodeRequestFunc will have a nil body.

-1

transport/http/client_test.go less more

28	28	client := httptransport.Client{
29	29	Method: "GET",
30	30	URL: mustParse(server.URL),
31		Context: context.Background(),
32	31	EncodeRequestFunc: encode,
33	32	DecodeResponseFunc: decode,
34	33	Before: []httptransport.RequestFunc{httptransport.SetRequestHeader(headerKey, headerVal)},

-0

transport/http/server.go less more

5	5	"golang.org/x/net/context"
6	6
7	7	"github.com/go-kit/kit/endpoint"
	8	"github.com/go-kit/kit/log"
8	9	)
9	10
10	11	// Server wraps an endpoint and implements http.Handler.

35	36	// ErrorEncoder is nil, the error will be written as plain text with
36	37	// an appropriate, if generic, status code.
37	38	ErrorEncoder func(w http.ResponseWriter, err error)
	39
	40	// Logger is used to log errors.
	41	Logger log.Logger
38	42	}
39	43
40	44	// ServeHTTP implements http.Handler.

52	56
53	57	request, err := s.DecodeRequestFunc(r)
54	58	if err != nil {
	59	_ = s.Logger.Log("err", err)
55	60	s.ErrorEncoder(w, badRequestError{err})
56	61	return
57	62	}
58	63
59	64	response, err := s.Endpoint(ctx, request)
60	65	if err != nil {
	66	_ = s.Logger.Log("err", err)
61	67	s.ErrorEncoder(w, err)
62	68	return
63	69	}

67	73	}
68	74
69	75	if err := s.EncodeResponseFunc(w, response); err != nil {
	76	_ = s.Logger.Log("err", err)
70	77	s.ErrorEncoder(w, err)
71	78	return
72	79	}

-0

transport/http/server_test.go less more

8	8
9	9	"golang.org/x/net/context"
10	10
	11	"github.com/go-kit/kit/log"
11	12	httptransport "github.com/go-kit/kit/transport/http"
12	13	)
13	14

17	18	Endpoint: func(context.Context, interface{}) (interface{}, error) { return struct{}{}, nil },
18	19	DecodeRequestFunc: func(*http.Request) (interface{}, error) { return struct{}{}, errors.New("dang") },
19	20	EncodeResponseFunc: func(http.ResponseWriter, interface{}) error { return nil },
	21	Logger: log.NewNopLogger(),
20	22	}
21	23	server := httptest.NewServer(handler)
22	24	defer server.Close()

32	34	Endpoint: func(context.Context, interface{}) (interface{}, error) { return struct{}{}, errors.New("dang") },
33	35	DecodeRequestFunc: func(*http.Request) (interface{}, error) { return struct{}{}, nil },
34	36	EncodeResponseFunc: func(http.ResponseWriter, interface{}) error { return nil },
	37	Logger: log.NewNopLogger(),
35	38	}
36	39	server := httptest.NewServer(handler)
37	40	defer server.Close()

47	50	Endpoint: func(context.Context, interface{}) (interface{}, error) { return struct{}{}, nil },
48	51	DecodeRequestFunc: func(*http.Request) (interface{}, error) { return struct{}{}, nil },
49	52	EncodeResponseFunc: func(http.ResponseWriter, interface{}) error { return errors.New("dang") },
	53	Logger: log.NewNopLogger(),
50	54	}
51	55	server := httptest.NewServer(handler)
52	56	defer server.Close()

70	74	DecodeRequestFunc: func(*http.Request) (interface{}, error) { return struct{}{}, nil },
71	75	EncodeResponseFunc: func(http.ResponseWriter, interface{}) error { return nil },
72	76	ErrorEncoder: func(w http.ResponseWriter, err error) { w.WriteHeader(code(err)) },
	77	Logger: log.NewNopLogger(),
73	78	}
74	79	server := httptest.NewServer(handler)
75	80	defer server.Close()

103	108	EncodeResponseFunc: func(http.ResponseWriter, interface{}) error { return nil },
104	109	Before: []httptransport.RequestFunc{func(ctx context.Context, r *http.Request) context.Context { return ctx }},
105	110	After: []httptransport.ResponseFunc{func(ctx context.Context, w http.ResponseWriter) { return }},
	111	Logger: log.NewNopLogger(),
106	112	}
107	113	)
108	114	go func() {