Migrate away from queue type in faas project

The queue type now resides in the provider, so that there is
no risk of a circular reference.

Signed-off-by: Alex Ellis (OpenFaaS Ltd) <alexellis2@gmail.com>
This commit is contained in:
Alex Ellis (OpenFaaS Ltd)
2021-07-26 17:21:20 +01:00
committed by Alex Ellis
parent 06a51373e2
commit 58394bb1de
101 changed files with 10232 additions and 1763 deletions

View File

@ -13,6 +13,7 @@ import (
"strings"
"sync"
"google.golang.org/protobuf/reflect/protodesc"
"google.golang.org/protobuf/reflect/protoreflect"
"google.golang.org/protobuf/reflect/protoregistry"
"google.golang.org/protobuf/runtime/protoimpl"
@ -62,14 +63,7 @@ func FileDescriptor(s filePath) fileDescGZIP {
// Find the descriptor in the v2 registry.
var b []byte
if fd, _ := protoregistry.GlobalFiles.FindFileByPath(s); fd != nil {
if fd, ok := fd.(interface{ ProtoLegacyRawDesc() []byte }); ok {
b = fd.ProtoLegacyRawDesc()
} else {
// TODO: Use protodesc.ToFileDescriptorProto to construct
// a descriptorpb.FileDescriptorProto and marshal it.
// However, doing so causes the proto package to have a dependency
// on descriptorpb, leading to cyclic dependency issues.
}
b, _ = Marshal(protodesc.ToFileDescriptorProto(fd))
}
// Locally cache the raw descriptor form for the file.

View File

@ -19,6 +19,8 @@ const urlPrefix = "type.googleapis.com/"
// AnyMessageName returns the message name contained in an anypb.Any message.
// Most type assertions should use the Is function instead.
//
// Deprecated: Call the any.MessageName method instead.
func AnyMessageName(any *anypb.Any) (string, error) {
name, err := anyMessageName(any)
return string(name), err
@ -38,6 +40,8 @@ func anyMessageName(any *anypb.Any) (protoreflect.FullName, error) {
}
// MarshalAny marshals the given message m into an anypb.Any message.
//
// Deprecated: Call the anypb.New function instead.
func MarshalAny(m proto.Message) (*anypb.Any, error) {
switch dm := m.(type) {
case DynamicAny:
@ -58,6 +62,9 @@ func MarshalAny(m proto.Message) (*anypb.Any, error) {
// Empty returns a new message of the type specified in an anypb.Any message.
// It returns protoregistry.NotFound if the corresponding message type could not
// be resolved in the global registry.
//
// Deprecated: Use protoregistry.GlobalTypes.FindMessageByName instead
// to resolve the message name and create a new instance of it.
func Empty(any *anypb.Any) (proto.Message, error) {
name, err := anyMessageName(any)
if err != nil {
@ -76,6 +83,8 @@ func Empty(any *anypb.Any) (proto.Message, error) {
//
// The target message m may be a *DynamicAny message. If the underlying message
// type could not be resolved, then this returns protoregistry.NotFound.
//
// Deprecated: Call the any.UnmarshalTo method instead.
func UnmarshalAny(any *anypb.Any, m proto.Message) error {
if dm, ok := m.(*DynamicAny); ok {
if dm.Message == nil {
@ -100,6 +109,8 @@ func UnmarshalAny(any *anypb.Any, m proto.Message) error {
}
// Is reports whether the Any message contains a message of the specified type.
//
// Deprecated: Call the any.MessageIs method instead.
func Is(any *anypb.Any, m proto.Message) bool {
if any == nil || m == nil {
return false
@ -119,6 +130,9 @@ func Is(any *anypb.Any, m proto.Message) bool {
// var x ptypes.DynamicAny
// if err := ptypes.UnmarshalAny(a, &x); err != nil { ... }
// fmt.Printf("unmarshaled message: %v", x.Message)
//
// Deprecated: Use the any.UnmarshalNew method instead to unmarshal
// the any message contents into a new instance of the underlying message.
type DynamicAny struct{ proto.Message }
func (m DynamicAny) String() string {

View File

@ -3,4 +3,8 @@
// license that can be found in the LICENSE file.
// Package ptypes provides functionality for interacting with well-known types.
//
// Deprecated: Well-known types have specialized functionality directly
// injected into the generated packages for each message type.
// See the deprecation notice for each function for the suggested alternative.
package ptypes

View File

@ -21,6 +21,8 @@ const (
// Duration converts a durationpb.Duration to a time.Duration.
// Duration returns an error if dur is invalid or overflows a time.Duration.
//
// Deprecated: Call the dur.AsDuration and dur.CheckValid methods instead.
func Duration(dur *durationpb.Duration) (time.Duration, error) {
if err := validateDuration(dur); err != nil {
return 0, err
@ -39,6 +41,8 @@ func Duration(dur *durationpb.Duration) (time.Duration, error) {
}
// DurationProto converts a time.Duration to a durationpb.Duration.
//
// Deprecated: Call the durationpb.New function instead.
func DurationProto(d time.Duration) *durationpb.Duration {
nanos := d.Nanoseconds()
secs := nanos / 1e9

View File

@ -33,6 +33,8 @@ const (
//
// A nil Timestamp returns an error. The first return value in that case is
// undefined.
//
// Deprecated: Call the ts.AsTime and ts.CheckValid methods instead.
func Timestamp(ts *timestamppb.Timestamp) (time.Time, error) {
// Don't return the zero value on error, because corresponds to a valid
// timestamp. Instead return whatever time.Unix gives us.
@ -46,6 +48,8 @@ func Timestamp(ts *timestamppb.Timestamp) (time.Time, error) {
}
// TimestampNow returns a google.protobuf.Timestamp for the current time.
//
// Deprecated: Call the timestamppb.Now function instead.
func TimestampNow() *timestamppb.Timestamp {
ts, err := TimestampProto(time.Now())
if err != nil {
@ -56,6 +60,8 @@ func TimestampNow() *timestamppb.Timestamp {
// TimestampProto converts the time.Time to a google.protobuf.Timestamp proto.
// It returns an error if the resulting Timestamp is invalid.
//
// Deprecated: Call the timestamppb.New function instead.
func TimestampProto(t time.Time) (*timestamppb.Timestamp, error) {
ts := &timestamppb.Timestamp{
Seconds: t.Unix(),
@ -69,6 +75,9 @@ func TimestampProto(t time.Time) (*timestamppb.Timestamp, error) {
// TimestampString returns the RFC 3339 string for valid Timestamps.
// For invalid Timestamps, it returns an error message in parentheses.
//
// Deprecated: Call the ts.AsTime method instead,
// followed by a call to the Format method on the time.Time value.
func TimestampString(ts *timestamppb.Timestamp) string {
t, err := Timestamp(ts)
if err != nil {

View File

@ -2,6 +2,22 @@ package types
import "time"
// Secret for underlying orchestrator
type Secret struct {
// Name of the secret
Name string `json:"name"`
// Namespace if applicable for the secret
Namespace string `json:"namespace,omitempty"`
// Value is a string representing the string's value
Value string `json:"value,omitempty"`
// RawValue can be used to provide binary data when
// Value is not set
RawValue []byte `json:"rawValue,omitempty"`
}
// FunctionDeployment represents a request to create or update a Function.
type FunctionDeployment struct {
@ -46,13 +62,6 @@ type FunctionDeployment struct {
ReadOnlyRootFilesystem bool `json:"readOnlyRootFilesystem,omitempty"`
}
// Secret for underlying orchestrator
type Secret struct {
Name string `json:"name"`
Namespace string `json:"namespace,omitempty"`
Value string `json:"value,omitempty"`
}
// FunctionResources Memory and CPU
type FunctionResources struct {
Memory string `json:"memory,omitempty"`

View File

@ -0,0 +1,42 @@
package types
import (
"net/http"
"net/url"
)
// Request for asynchronous processing
type QueueRequest struct {
// Header from HTTP request
Header http.Header
// Host from HTTP request
Host string
// Body from HTTP request to use for invocation
Body []byte
// Method from HTTP request
Method string
// Path from HTTP request
Path string
// QueryString from HTTP request
QueryString string
// Function name to invoke
Function string
// QueueName to publish the request to, leave blank
// for default.
QueueName string
// Used by queue worker to submit a result
CallbackURL *url.URL `json:"CallbackUrl"`
}
// RequestQueuer can public a request to be executed asynchronously
type RequestQueuer interface {
Queue(req *QueueRequest) error
}

View File

@ -8,7 +8,7 @@ import (
"time"
stan "github.com/nats-io/stan.go"
"github.com/openfaas/faas/gateway/queue"
ftypes "github.com/openfaas/faas-provider/types"
)
// NATSQueue queue for work
@ -32,7 +32,7 @@ type NATSQueue struct {
}
// Queue request for processing
func (q *NATSQueue) Queue(req *queue.Request) error {
func (q *NATSQueue) Queue(req *ftypes.QueueRequest) error {
fmt.Printf("NatsQueue - submitting request: %s.\n", req.Function)
out, err := json.Marshal(req)