v2.1.0: add MongoDB package. (#17)

* feat: add mongo_client package

* feat: add mongo_client package

* test: add mongo_client package tests

* remove binary

* style fix

* remove mongo entrypoint

* add build script

* refactor config

* refactor config mechanism

* change default port

* delete go sum

* add go doc

* fix bad naming

* remove useless bin

* fix flag

* change config priority

* delay flag parsing

* delay flag parsing

* reflect only print to stdout

.gitignore

@@ -4,3 +4,7 @@ composer.lock
 *.log
 .idea/
 /app
+/mongo
+/example/app
+/example/go2php/app
+/example/config/app

.travis.yml

@@ -5,18 +5,14 @@ sudo: required
 matrix:
   include:
     - php: 7.2
-      env: SW_VERSION="4.5.0"
+      env: SW_VERSION="4.5.2"
     - php: 7.3
-      env: SW_VERSION="4.5.0"
+      env: SW_VERSION="4.5.2"
     - php: 7.4
-      env: SW_VERSION="4.5.0"
-    - php: 7.2
-      env: SW_VERSION="4.4.18"
-    - php: 7.3
-      env: SW_VERSION="4.4.18"
-    - php: 7.4
-      env: SW_VERSION="4.4.18"
-services: {}
+      env: SW_VERSION="4.5.2"
+
+services:
+  - mongodb
 
 before_install:
   - export PHP_MAJOR="$(`phpenv which php` -r 'echo phpversion();' | cut -d '.' -f 1)"

bin/build.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+package=../example/mongo_client/sidecar.go
+package_name=mongo-proxy
+
+#the full list of the platforms: https://golang.org/doc/install/source#environment
+platforms=(
+"darwin/amd64"
+"linux/amd64"
+)
+
+for platform in "${platforms[@]}"
+do
+    platform_split=(${platform//\// })
+    GOOS=${platform_split[0]}
+    GOARCH=${platform_split[1]}
+    output_name=$package_name'-'$GOOS'-'$GOARCH
+    if [ $GOOS = "windows" ]; then
+        output_name+='.exe'
+    fi
+    echo GOOS=$GOOS GOARCH=$GOARCH go build -o $output_name $package
+    GOOS=$GOOS GOARCH=$GOARCH go build -o $output_name $package
+    if [ $? -ne 0 ]; then
+        echo 'An error has occurred! Aborting the script execution...'
+        exit 1
+    fi
+done

composer.json

@@ -26,22 +26,24 @@
     },
     "require-dev": {
         "friendsofphp/php-cs-fixer": "^2.14",
+        "hyperf/command": "^1.1",
         "hyperf/config": "^1.1",
         "hyperf/di": "^1.1",
         "hyperf/testing": "1.1.*",
         "mockery/mockery": "^1.3",
         "phpstan/phpstan": "^0.12",
-        "swoft/swoole-ide-helper": "dev-master"
+        "swoole/ide-helper": "^4.5"
     },
     "config": {
         "sort-packages": true
     },
     "scripts": {
-        "test": "go build -o app example/*.go && phpunit -c phpunit.xml --colors=always",
+        "test": "go build -o app example/*.go && go build -o mongo example/mongo_client/*.go && phpunit -c phpunit.xml --colors=always",
         "start-test-server": "php tests/TestServer.php",
         "test-go": "/bin/bash -c 'php tests/TestServer.php & sleep 5 && go test ./...'",
         "analyse": "phpstan analyse --memory-limit 300M -l 0 ./src",
-        "cs-fix": "php-cs-fixer fix $1"
+        "cs-fix": "php-cs-fixer fix $1",
+        "binary": "go build -o mongo example/mongo_client/*.go"
     },
     "extra": {
         "hyperf": {

example/main.php

@@ -9,7 +9,6 @@
  * @contact  guxi99@gmail.com
  * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
  */
-
 use Hyperf\GoTask\GoTask;
 use Hyperf\GoTask\IPC\SocketIPCSender;
 use Swoole\Process;

example/mongo_client/main.php

@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf/GoTask.
+ *
+ * @link     https://www.github.com/hyperf/gotask
+ * @document  https://www.github.com/hyperf/gotask
+ * @contact  guxi99@gmail.com
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+use Reasno\GoTask\GoTask;
+use Reasno\GoTask\IPC\SocketIPCSender;
+use Swoole\Process;
+use function Swoole\Coroutine\run;
+
+require __DIR__ . '/../../vendor/autoload.php';
+
+const ADDR = '127.0.0.1:6001';
+
+exec('go build -o ' . __DIR__ . '/app ' . __DIR__ . '/sidecar.go');
+$process = new Process(function (Process $process) {
+    $process->exec(__DIR__ . '/app', ['-address', ADDR]);
+});
+$process->start();
+
+sleep(1);
+
+run(function () {
+    $task = new SocketIPCSender(ADDR);
+    for ($i = 0; $i < 5; ++$i) {
+        $task->call('MongoProxy.InsertOne', ['Database' => 'testing', 'Collection' => 'colors', 'Record' => ['Blue' => 'Red', 'number' => $i]]);
+    }
+    var_dump($task->call('MongoProxy.Find', ['Database' => 'testing', 'Collection' => 'colors', 'Filter' => ['Blue' => 'Red'], 'Opts' => [['Skip' => 1, 'Limit' => 2]]]));
+});

example/mongo_client/sidecar.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"context"
+	"log"
+
+	"github.com/hyperf/gotask/v2/pkg/gotask"
+	"github.com/hyperf/gotask/v2/pkg/mongo_client"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func main() {
+	mongoConfig := mongo_client.LoadConfig()
+	ctx, cancel := context.WithTimeout(context.Background(), mongoConfig.ConnectTimeout)
+	defer cancel()
+
+	client, err := mongo.Connect(ctx, options.Client().ApplyURI(mongoConfig.Uri))
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	if err := gotask.Register(mongo_client.NewMongoProxyWithTimeout(client, mongoConfig.ReadWriteTimeout)); err != nil {
+		log.Fatalln(err)
+	}
+
+	if err := gotask.Run(); err != nil {
+		log.Fatalln(err)
+	}
+}

go.mod

@@ -4,8 +4,9 @@ go 1.13
 
 require (
 	github.com/fatih/pool v3.0.0+incompatible
-	github.com/json-iterator/go v1.1.9
 	github.com/oklog/run v1.1.0
 	github.com/pkg/errors v0.9.1
+	github.com/reasno/gotask v1.0.2
 	github.com/spiral/goridge/v2 v2.3.3
+	go.mongodb.org/mongo-driver v1.3.3
 )

pkg/gotask/flag.go

@@ -2,7 +2,6 @@ package gotask
 
 import (
 	"flag"
-	"testing"
 )
 
 var (
@@ -14,11 +13,9 @@ var (
 )
 
 func init() {
-	testing.Init()
 	standalone = flag.Bool("standalone", false, "if set, ignore parent process status")
 	address = flag.String("address", "127.0.0.1:6001", "must be a unix socket or tcp address:port like 127.0.0.1:6001")
 	listenOnPipe = flag.Bool("listen-on-pipe", false, "listen on stdin/stdout pipe")
 	go2phpAddress = flag.String("go2php-address", "127.0.0.1:6002", "must be a unix socket or tcp address:port like 127.0.0.1:6002")
 	reflection = flag.Bool("reflect", false, "instead of running the service, provide a service definition to os.stdout using reflection")
-	flag.Parse()
 }

pkg/gotask/generator.go

@@ -2,6 +2,7 @@ package gotask
 
 import (
 	"bytes"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -105,19 +106,24 @@ func body(namespace *string, class *Class) string {
 func generatePHP(receiver interface{}) error {
 	namespace := property(receiver, "PHPNamespace", "App\\GoTask")
 	class := reflectStruct(receiver)
-	dirPath := property(receiver, "PHPPath", "./../app/GoTask")
-	err := os.MkdirAll(dirPath, os.FileMode(0755))
-	if err != nil {
-		return errors.Wrap(err, "cannot create dir for php files")
-	}
-	fullPath, err := filepath.Abs(filepath.Clean(dirPath) + "/" + class.Name + ".php")
-	if err != nil {
-		return errors.Wrap(err, "invalid file path")
-	}
 	out := body(&namespace, class)
-	err = ioutil.WriteFile(fullPath, []byte(out), os.FileMode(0755))
-	if err != nil {
-		return errors.Wrap(err, "failed to generate PHP file")
+	dirPath := property(receiver, "PHPPath", "")
+	if dirPath != "" {
+		err := os.MkdirAll(dirPath, os.FileMode(0755))
+		if err != nil {
+			return errors.Wrap(err, "cannot create dir for php files")
+		}
+		fullPath, err := filepath.Abs(filepath.Clean(dirPath) + "/" + class.Name + ".php")
+		if err != nil {
+			return errors.Wrap(err, "invalid file path")
+		}
+
+		err = ioutil.WriteFile(fullPath, []byte(out), os.FileMode(0755))
+		if err != nil {
+			return errors.Wrap(err, "failed to generate PHP file")
+		}
+	} else {
+		fmt.Print(out)
 	}
 	return nil
 }

pkg/gotask/server.go

@@ -2,16 +2,18 @@ package gotask
 
 import (
 	"context"
+	"flag"
 	"fmt"
-	"github.com/oklog/run"
-	"github.com/pkg/errors"
-	"github.com/spiral/goridge/v2"
 	"net"
 	"net/rpc"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
+
+	"github.com/oklog/run"
+	"github.com/pkg/errors"
+	"github.com/spiral/goridge/v2"
 )
 
 var g run.Group
@@ -33,6 +35,9 @@ func checkProcess(pid int, quit chan bool) {
 
 // Register a net/rpc compatible service
 func Register(receiver interface{}) error {
+	if !flag.Parsed() {
+		flag.Parse()
+	}
 	if !*reflection {
 		return rpc.Register(receiver)
 	}
@@ -51,6 +56,10 @@ func GetAddress() string {
 
 // Run the sidecar, receive any fatal errors.
 func Run() error {
+	if !flag.Parsed() {
+		flag.Parse()
+	}
+
 	if *reflection {
 		return nil
 	}

pkg/mongo_client/config.go

@@ -0,0 +1,55 @@
+package mongo_client
+
+import (
+	"flag"
+	"os"
+	"time"
+)
+
+type Config struct {
+	Uri              string
+	ConnectTimeout   time.Duration
+	ReadWriteTimeout time.Duration
+}
+
+var (
+	globalMongoUri              *string
+	globalMongoConnectTimeout   *time.Duration
+	globalMongoReadWriteTimeout *time.Duration
+)
+
+func init() {
+	uri, ok := os.LookupEnv("MONGODB_URI")
+	if !ok {
+		uri = "mongodb://127.0.0.1:27017"
+	}
+	ct := getTimeout("MONGODB_CONNECT_TIMEOUT", 3*time.Second)
+	rwt := getTimeout("MONGODB_READ_WRITE_TIMEOUT", time.Minute)
+
+	globalMongoUri = flag.String("mongodb-uri", uri, "the default mongodb uri")
+	globalMongoConnectTimeout = flag.Duration("mongodb-connect-timeout", ct, "mongodb connect timeout")
+	globalMongoReadWriteTimeout = flag.Duration("mongodb-read-write-timeout", rwt, "mongodb read write timeout")
+	flag.Parse()
+}
+
+func getTimeout(env string, fallback time.Duration) (result time.Duration) {
+	env, ok := os.LookupEnv(env)
+	if !ok {
+		return fallback
+	}
+	result, err := time.ParseDuration(env)
+	if err != nil {
+		return fallback
+	}
+	return result
+}
+
+// LoadConfig loads Configurations from environmental variables or config file in PHP.
+// Environmental variables takes priority.
+func LoadConfig() Config {
+	return Config{
+		*globalMongoUri,
+		*globalMongoConnectTimeout,
+		*globalMongoReadWriteTimeout,
+	}
+}