[v3] fix panic on publish methods (#167)
* move publish methods to a separate file * reorder client methods * fix panic for publish * refactoring * go mod tidy * rename go.micro => micro * add comment to README about Set-Cookie headers
This commit is contained in:
		| @@ -13,6 +13,7 @@ implements HTTP rules defined in the [google/api/http.proto](https://github.com/ | ||||
|  | ||||
| * Streaming is not yet implemented. | ||||
| * Only protobuf-generated messages are supported. | ||||
| * In `micro/v3`, metadata is implemented as `map[string]string`, which works for most headers but not for multiple `Set-Cookie` headers. The HTTP specification forbids the use of commas in `Set-Cookie` headers; therefore, their values cannot be parsed reliably. In `micro/v4`, metadata uses `map[string][]string`, resolving this issue. | ||||
|  | ||||
| ## Usage | ||||
|  | ||||
|   | ||||
							
								
								
									
										109
									
								
								client.go
									
									
									
									
									
								
							
							
						
						
									
										109
									
								
								client.go
									
									
									
									
									
								
							| @@ -3,20 +3,17 @@ package http | ||||
| import ( | ||||
| 	"context" | ||||
| 	"net/http" | ||||
| 	"os" | ||||
| 	"strconv" | ||||
| 	"sync" | ||||
| 	"time" | ||||
|  | ||||
| 	"go.unistack.org/micro-client-http/v3/status" | ||||
| 	"go.unistack.org/micro/v3/broker" | ||||
| 	"go.unistack.org/micro/v3/client" | ||||
| 	"go.unistack.org/micro/v3/codec" | ||||
| 	"go.unistack.org/micro/v3/errors" | ||||
| 	"go.unistack.org/micro/v3/metadata" | ||||
| 	"go.unistack.org/micro/v3/options" | ||||
| 	"go.unistack.org/micro/v3/semconv" | ||||
| 	"go.unistack.org/micro/v3/tracer" | ||||
|  | ||||
| 	"go.unistack.org/micro-client-http/v3/status" | ||||
| ) | ||||
|  | ||||
| var _ client.Client = (*Client)(nil) | ||||
| @@ -52,6 +49,8 @@ func NewClient(opts ...client.Option) *Client { | ||||
| 		c.httpClient = defaultHTTPClient(dialer, clientOpts.TLSConfig) | ||||
| 	} | ||||
|  | ||||
| 	c.funcPublish = c.fnPublish | ||||
| 	c.funcBatchPublish = c.fnBatchPublish | ||||
| 	c.funcCall = c.fnCall | ||||
| 	c.funcStream = c.fnStream | ||||
|  | ||||
| @@ -73,6 +72,10 @@ func (c *Client) Init(opts ...client.Option) error { | ||||
| 			c.funcCall = h(c.funcCall) | ||||
| 		case client.HookStream: | ||||
| 			c.funcStream = h(c.funcStream) | ||||
| 		case client.HookPublish: | ||||
| 			c.funcPublish = h(c.funcPublish) | ||||
| 		case client.HookBatchPublish: | ||||
| 			c.funcBatchPublish = h(c.funcBatchPublish) | ||||
| 		} | ||||
| 	}) | ||||
|  | ||||
| @@ -83,6 +86,20 @@ func (c *Client) Options() client.Options { | ||||
| 	return c.opts | ||||
| } | ||||
|  | ||||
| func (c *Client) NewMessage(topic string, msg interface{}, opts ...client.MessageOption) client.Message { | ||||
| 	msgOpts := client.NewMessageOptions(opts...) | ||||
| 	if msgOpts.ContentType == "" { | ||||
| 		msgOpts.ContentType = c.opts.ContentType | ||||
| 	} | ||||
|  | ||||
| 	return &httpMessage{ | ||||
| 		topic:   topic, | ||||
| 		payload: msg, | ||||
| 		opts:    msgOpts, | ||||
| 	} | ||||
|  | ||||
| } | ||||
|  | ||||
| func (c *Client) NewRequest(service, method string, req any, opts ...client.RequestOption) client.Request { | ||||
| 	reqOpts := client.NewRequestOptions(opts...) | ||||
| 	if reqOpts.ContentType == "" { | ||||
| @@ -146,88 +163,14 @@ func (c *Client) Stream(ctx context.Context, req client.Request, opts ...client. | ||||
| 	return c.funcStream(ctx, req, opts...) | ||||
| } | ||||
|  | ||||
| func (c *Client) String() string { | ||||
| 	return "http" | ||||
| func (c *Client) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error { | ||||
| 	return c.funcPublish(ctx, p, opts...) | ||||
| } | ||||
|  | ||||
| func (c *Client) BatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error { | ||||
| 	return c.funcBatchPublish(ctx, ps, opts...) | ||||
| } | ||||
|  | ||||
| func (c *Client) fnBatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error { | ||||
| 	return c.publish(ctx, ps, opts...) | ||||
| } | ||||
|  | ||||
| func (c *Client) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error { | ||||
| 	return c.funcPublish(ctx, p, opts...) | ||||
| } | ||||
|  | ||||
| func (c *Client) fnPublish(ctx context.Context, p client.Message, opts ...client.PublishOption) error { | ||||
| 	return c.publish(ctx, []client.Message{p}, opts...) | ||||
| } | ||||
|  | ||||
| func (c *Client) publish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error { | ||||
| 	var body []byte | ||||
|  | ||||
| 	options := client.NewPublishOptions(opts...) | ||||
|  | ||||
| 	// get proxy | ||||
| 	exchange := "" | ||||
| 	if v, ok := os.LookupEnv("MICRO_PROXY"); ok { | ||||
| 		exchange = v | ||||
| 	} | ||||
| 	// get the exchange | ||||
| 	if len(options.Exchange) > 0 { | ||||
| 		exchange = options.Exchange | ||||
| 	} | ||||
|  | ||||
| 	msgs := make([]*broker.Message, 0, len(ps)) | ||||
|  | ||||
| 	omd, ok := metadata.FromOutgoingContext(ctx) | ||||
| 	if !ok { | ||||
| 		omd = metadata.New(2) | ||||
| 	} | ||||
|  | ||||
| 	for _, p := range ps { | ||||
| 		md := metadata.Copy(omd) | ||||
| 		topic := p.Topic() | ||||
| 		if len(exchange) > 0 { | ||||
| 			topic = exchange | ||||
| 		} | ||||
| 		md.Set(metadata.HeaderTopic, topic) | ||||
| 		iter := p.Metadata().Iterator() | ||||
| 		var k, v string | ||||
| 		for iter.Next(&k, &v) { | ||||
| 			md.Set(k, v) | ||||
| 		} | ||||
|  | ||||
| 		md[metadata.HeaderContentType] = p.ContentType() | ||||
|  | ||||
| 		// passed in raw data | ||||
| 		if d, ok := p.Payload().(*codec.Frame); ok { | ||||
| 			body = d.Data | ||||
| 		} else { | ||||
| 			// use codec for payload | ||||
| 			cf, err := c.newCodec(p.ContentType()) | ||||
| 			if err != nil { | ||||
| 				return errors.InternalServerError("go.micro.client", "%+v", err) | ||||
| 			} | ||||
| 			// set the body | ||||
| 			b, err := cf.Marshal(p.Payload()) | ||||
| 			if err != nil { | ||||
| 				return errors.InternalServerError("go.micro.client", "%+v", err) | ||||
| 			} | ||||
| 			body = b | ||||
| 		} | ||||
| 		msgs = append(msgs, &broker.Message{Header: md, Body: body}) | ||||
| 	} | ||||
|  | ||||
| 	return c.opts.Broker.BatchPublish(ctx, msgs, | ||||
| 		broker.PublishContext(options.Context), | ||||
| 		broker.PublishBodyOnly(options.BodyOnly), | ||||
| 	) | ||||
| } | ||||
|  | ||||
| func (c *Client) NewMessage(topic string, msg interface{}, opts ...client.MessageOption) client.Message { | ||||
| 	return newHTTPEvent(topic, msg, c.opts.ContentType, opts...) | ||||
| func (c *Client) String() string { | ||||
| 	return "http" | ||||
| } | ||||
|   | ||||
							
								
								
									
										84
									
								
								client_publish.go
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										84
									
								
								client_publish.go
									
									
									
									
									
										Normal file
									
								
							| @@ -0,0 +1,84 @@ | ||||
| package http | ||||
|  | ||||
| import ( | ||||
| 	"context" | ||||
| 	"os" | ||||
|  | ||||
| 	"go.unistack.org/micro/v3/broker" | ||||
| 	"go.unistack.org/micro/v3/client" | ||||
| 	"go.unistack.org/micro/v3/codec" | ||||
| 	"go.unistack.org/micro/v3/errors" | ||||
| 	"go.unistack.org/micro/v3/metadata" | ||||
| ) | ||||
|  | ||||
| func (c *Client) publish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error { | ||||
| 	var body []byte | ||||
|  | ||||
| 	options := client.NewPublishOptions(opts...) | ||||
|  | ||||
| 	// get proxy | ||||
| 	exchange := "" | ||||
| 	if v, ok := os.LookupEnv("MICRO_PROXY"); ok { | ||||
| 		exchange = v | ||||
| 	} | ||||
| 	// get the exchange | ||||
| 	if len(options.Exchange) > 0 { | ||||
| 		exchange = options.Exchange | ||||
| 	} | ||||
|  | ||||
| 	msgs := make([]*broker.Message, 0, len(ps)) | ||||
|  | ||||
| 	omd, ok := metadata.FromOutgoingContext(ctx) | ||||
| 	if !ok { | ||||
| 		omd = metadata.New(2) | ||||
| 	} | ||||
|  | ||||
| 	for _, p := range ps { | ||||
| 		md := metadata.Copy(omd) | ||||
| 		topic := p.Topic() | ||||
| 		if len(exchange) > 0 { | ||||
| 			topic = exchange | ||||
| 		} | ||||
| 		md.Set(metadata.HeaderTopic, topic) | ||||
| 		iter := p.Metadata().Iterator() | ||||
| 		var k, v string | ||||
| 		for iter.Next(&k, &v) { | ||||
| 			md.Set(k, v) | ||||
| 		} | ||||
|  | ||||
| 		md[metadata.HeaderContentType] = p.ContentType() | ||||
|  | ||||
| 		// passed in raw data | ||||
| 		if d, ok := p.Payload().(*codec.Frame); ok { | ||||
| 			body = d.Data | ||||
| 		} else { | ||||
| 			// use codec for payload | ||||
| 			cf, err := c.newCodec(p.ContentType()) | ||||
| 			if err != nil { | ||||
| 				return errors.InternalServerError("go.micro.client", "%+v", err) | ||||
| 			} | ||||
| 			// set the body | ||||
| 			b, err := cf.Marshal(p.Payload()) | ||||
| 			if err != nil { | ||||
| 				return errors.InternalServerError("go.micro.client", "%+v", err) | ||||
| 			} | ||||
| 			body = b | ||||
| 		} | ||||
| 		msgs = append(msgs, &broker.Message{Header: md, Body: body}) | ||||
| 	} | ||||
|  | ||||
| 	return c.opts.Broker.BatchPublish( | ||||
| 		ctx, | ||||
| 		msgs, | ||||
| 		broker.PublishContext(options.Context), | ||||
| 		broker.PublishBodyOnly(options.BodyOnly), | ||||
| 	) | ||||
| } | ||||
|  | ||||
| func (c *Client) fnPublish(ctx context.Context, p client.Message, opts ...client.PublishOption) error { | ||||
| 	return c.publish(ctx, []client.Message{p}, opts...) | ||||
| } | ||||
|  | ||||
| func (c *Client) fnBatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error { | ||||
| 	return c.publish(ctx, ps, opts...) | ||||
| } | ||||
							
								
								
									
										2
									
								
								go.mod
									
									
									
									
									
								
							
							
						
						
									
										2
									
								
								go.mod
									
									
									
									
									
								
							| @@ -7,7 +7,7 @@ toolchain go1.24.3 | ||||
| require ( | ||||
| 	github.com/stretchr/testify v1.11.1 | ||||
| 	go.unistack.org/micro-codec-json/v3 v3.10.3 | ||||
| 	go.unistack.org/micro/v3 v3.11.45 | ||||
| 	go.unistack.org/micro/v3 v3.11.48 | ||||
| 	google.golang.org/protobuf v1.36.10 | ||||
| ) | ||||
|  | ||||
|   | ||||
							
								
								
									
										4
									
								
								go.sum
									
									
									
									
									
								
							
							
						
						
									
										4
									
								
								go.sum
									
									
									
									
									
								
							| @@ -29,8 +29,8 @@ go.unistack.org/micro-codec-json/v3 v3.10.3 h1:FwSBfJswov30Dyqxp1XfQW1EG4h77uTEe | ||||
| go.unistack.org/micro-codec-json/v3 v3.10.3/go.mod h1:26OK5MizMNKhspGC6PRVwpDIp5w1GmRb0nE5eRWWDxA= | ||||
| go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q= | ||||
| go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo= | ||||
| go.unistack.org/micro/v3 v3.11.45 h1:fjTLZYWgsVf9FIMZBxOg8ios2/tmyimnjZrsrxEUeXU= | ||||
| go.unistack.org/micro/v3 v3.11.45/go.mod h1:fDQ8Mu9wubaFP0L8hNQlpzHiEnWN0wbOlawN9HYo0N4= | ||||
| go.unistack.org/micro/v3 v3.11.48 h1:lHJYSHU2z1TTcuswItGwG7cZXN6n04EFqY7lk/0gA7w= | ||||
| go.unistack.org/micro/v3 v3.11.48/go.mod h1:fDQ8Mu9wubaFP0L8hNQlpzHiEnWN0wbOlawN9HYo0N4= | ||||
| golang.org/x/net v0.41.0 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw= | ||||
| golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA= | ||||
| golang.org/x/sys v0.36.0 h1:KVRy2GtZBrk1cBYA7MKu5bEZFxQk4NIDV6RLVcC8o0k= | ||||
|   | ||||
							
								
								
									
										36
									
								
								message.go
									
									
									
									
									
								
							
							
						
						
									
										36
									
								
								message.go
									
									
									
									
									
								
							| @@ -5,40 +5,24 @@ import ( | ||||
| 	"go.unistack.org/micro/v3/metadata" | ||||
| ) | ||||
|  | ||||
| type httpEvent struct { | ||||
| 	payload     interface{} | ||||
| type httpMessage struct { | ||||
| 	topic   string | ||||
| 	contentType string | ||||
| 	payload interface{} | ||||
| 	opts    client.MessageOptions | ||||
| } | ||||
|  | ||||
| func newHTTPEvent(topic string, payload interface{}, contentType string, opts ...client.MessageOption) client.Message { | ||||
| 	options := client.NewMessageOptions(opts...) | ||||
|  | ||||
| 	if len(options.ContentType) > 0 { | ||||
| 		contentType = options.ContentType | ||||
| func (m *httpMessage) Topic() string { | ||||
| 	return m.topic | ||||
| } | ||||
|  | ||||
| 	return &httpEvent{ | ||||
| 		payload:     payload, | ||||
| 		topic:       topic, | ||||
| 		contentType: contentType, | ||||
| 		opts:        options, | ||||
| 	} | ||||
| func (m *httpMessage) Payload() interface{} { | ||||
| 	return m.payload | ||||
| } | ||||
|  | ||||
| func (c *httpEvent) ContentType() string { | ||||
| 	return c.contentType | ||||
| func (m *httpMessage) ContentType() string { | ||||
| 	return m.opts.ContentType | ||||
| } | ||||
|  | ||||
| func (c *httpEvent) Topic() string { | ||||
| 	return c.topic | ||||
| } | ||||
|  | ||||
| func (c *httpEvent) Payload() interface{} { | ||||
| 	return c.payload | ||||
| } | ||||
|  | ||||
| func (c *httpEvent) Metadata() metadata.Metadata { | ||||
| 	return c.opts.Metadata | ||||
| func (m *httpMessage) Metadata() metadata.Metadata { | ||||
| 	return m.opts.Metadata | ||||
| } | ||||
|   | ||||
		Reference in New Issue
	
	Block a user