add missing outgoing key to ctx for subscribers (#279 )

update ci (#276 )
disable coverage job (#274 )
2025-06-11 01:22:10 +03:00 · 2025-05-05 19:24:46 +03:00 · 2025-05-03 10:39:55 +03:00 · 2025-05-02 16:14:36 +00:00 · 2025-05-02 19:10:38 +03:00 · 2025-04-29 18:12:45 +03:00
22 changed files with 508 additions and 204 deletions
--- a/.github/ISSUE_TEMPLATE/bug_report.md
+++ b/.github/ISSUE_TEMPLATE/bug_report.md
@@ -1,6 +1,6 @@
 ---
 name: Bug report
-about: For reporting bugs in micro
+about: For reporting bugs in go-micro
 title: "[BUG]"
 labels: ''
 assignees: ''
@@ -16,3 +16,9 @@ assignees: ''
 **How to reproduce the bug:**

 If possible, please include a minimal code snippet here.
+
+**Environment:**
+Go Version: please paste `go version` output here
+```
+please paste `go env` output here
+```
--- a/.github/ISSUE_TEMPLATE/feature-request---enhancement.md
+++ b/.github/ISSUE_TEMPLATE/feature-request---enhancement.md
@@ -1,6 +1,6 @@
 ---
 name: Feature request / Enhancement
-about: If you have a need not served by micro
+about: If you have a need not served by go-micro
 title: "[FEATURE]"
 labels: ''
 assignees: ''
@@ -14,4 +14,4 @@ A clear and concise description of what the problem is. Ex. I'm always frustrate
 A clear and concise description of what you want to happen.

 **Additional context**
-Add any other context or screenshots about the feature request here.
+Add any other context or screenshots about the feature request here.
--- a/.github/ISSUE_TEMPLATE/question.md
+++ b/.github/ISSUE_TEMPLATE/question.md
@@ -1,8 +1,14 @@
 ---
 name: Question
-about: Ask a question about micro
+about: Ask a question about go-micro
 title: ''
 labels: ''
 assignees: ''

---
+---
+
+Before asking, please check if your question has already been answered: 
+
+1. Check the documentation - https://micro.mu/docs/
+2. Check the examples and plugins - https://github.com/micro/examples & https://github.com/micro/go-plugins
+3. Search existing issues
--- a/.github/autoapprove.yml
+++ b/.github/autoapprove.yml
@@ -1,28 +0,0 @@
-name: "autoapprove"
-
-on:
-  pull_request_target:
-    types: [assigned, opened, synchronize, reopened]
-  workflow_run:
-    workflows: ["prbuild"]
-    types:
-      - completed
-
-permissions:
-  pull-requests: write
-  contents: write
-
-jobs:
-  autoapprove:
-    runs-on: ubuntu-latest
-    steps:
-      - name: approve
-        run: [ "curl -o tea https://dl.gitea.com/tea/main/tea-main-linux-amd64",
-               "chmod +x ./tea",
-               "./tea login add --name unistack --token ${{ secrets.GITHUB_TOKEN }} --url https://git.unistack.org",
-               "./tea pr --repo ${{ github.event.repository.name }}"
-        ]
-        if: github.actor == 'vtolstov'
-        id: approve
-        with:
-          github-token: ${{ secrets.GITHUB_TOKEN }}
--- a/.github/workflows/job_coverage.yml
+++ b/.github/workflows/job_coverage.yml
@@ -8,8 +8,6 @@ on:
      - '.gitea/**'
  pull_request:
    branches: [ main, v3, v4 ]
-  # Allows you to run this workflow manually from the Actions tab
-  workflow_dispatch:

 jobs:

--- a/.github/workflows/job_lint.yml
+++ b/.github/workflows/job_lint.yml
@@ -20,7 +20,7 @@ jobs:
      uses: actions/setup-go@v5
      with:
        cache-dependency-path: "**/*.sum"
-        go-version: 'stable' 
+        go-version: 'stable'
    - name: setup deps
      run: go get -v ./...
    - name: run lint
--- a/.github/workflows/job_sync.yml
+++ b/.github/workflows/job_sync.yml
@@ -1,57 +0,0 @@
-name: sync
-
-on:
-  schedule:
-    - cron: '*/5 * * * *'
-#  push:
-#    branches: [ master, v3, v4 ]
-#    paths-ignore:
-#      - '.github/**'
-#      - '.gitea/**'
-  # Allows you to run this workflow manually from the Actions tab
-  workflow_dispatch:
-
-jobs:
-  sync:
-    if: github.server_url != 'https://github.com'
-    runs-on: ubuntu-latest
-    steps:
-    - name: init
-      run: |
-        git config --global user.email "vtolstov <vtolstov@users.noreply.github.com>"
-        git config --global user.name "github-actions[bot]"
-        echo "machine git.unistack.org login vtolstov password ${{ secrets.TOKEN_GITEA }}" >> /root/.netrc
-        echo "machine github.com login vtolstov password ${{ secrets.TOKEN_GITHUB }}" >> /root/.netrc
-
-    - name: sync master
-      run: |
-        git clone --filter=blob:none --filter=tree:0 --branch master --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
-        cd repo
-        git remote add --no-tags --track master upstream https://github.com/${GITHUB_REPOSITORY}
-        git pull --rebase upstream master
-        git push upstream master --progress
-        git push origin master --progress
-        cd ../
-        rm -rf repo
-
-    - name: sync v3
-      run: |
-        git clone --filter=blob:none --filter=tree:0 --branch v3 --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
-        cd repo
-        git remote add --no-tags --fetch --track v3 upstream https://github.com/${GITHUB_REPOSITORY}
-        git pull --rebase upstream v3
-        git push upstream v3
-        git push origin v3 --progress
-        cd ../
-        rm -rf repo
-
-    - name: sync v4
-      run: |
-        git clone --filter=blob:none --filter=tree:0 --branch v4 --single-branch ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY} repo
-        cd repo
-        git remote add --no-tags --fetch --track v4 upstream https://github.com/${GITHUB_REPOSITORY}
-        git pull --rebase upstream v4
-        git push upstream v4
-        git push origin v4 --progress
-        cd ../
-        rm -rf repo
--- a/README.md
+++ b/README.md
@@ -0,0 +1,4 @@
+# GRPC Server
+![Coverage](https://img.shields.io/badge/Coverage-2.6%25-red)
+
+This plugin is a grpc server for micro.
--- a/codec.go
+++ b/codec.go
@@ -1,7 +1,7 @@
 package grpc

 import (
-	"go.unistack.org/micro/v4/codec"
+	"go.unistack.org/micro/v3/codec"
 	"google.golang.org/grpc/encoding"
 )

--- a/error.go
+++ b/error.go
@@ -6,7 +6,7 @@ import (
 	"net/http"
 	"os"

-	"go.unistack.org/micro/v4/errors"
+	"go.unistack.org/micro/v3/errors"
 	"google.golang.org/grpc/codes"
 )

--- a/generate.go
+++ b/generate.go
@@ -1,4 +1,4 @@
 package grpc

 //go:generate go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
-//go:generate sh -c "protoc -I./proto -I$(go list -f '{{ .Dir }}' -m go.unistack.org/micro-proto/v4) -I. --go-grpc_out=paths=source_relative:./proto --go_out=paths=source_relative:./proto proto/test.proto"
+//go:generate sh -c "protoc -I./proto -I$(go list -f '{{ .Dir }}' -m go.unistack.org/micro-proto/v3) -I. --go-grpc_out=paths=source_relative:./proto --go_out=paths=source_relative:./proto proto/test.proto"
--- a/go.mod
+++ b/go.mod
@@ -1,24 +1,24 @@
-module go.unistack.org/micro-server-grpc/v4
+module go.unistack.org/micro-server-grpc/v3

-go 1.23.0
+go 1.22.2

 toolchain go1.24.2

 require (
-	go.unistack.org/micro/v4 v4.1.8
-	golang.org/x/net v0.39.0
-	google.golang.org/grpc v1.72.0
-	google.golang.org/protobuf v1.36.6
+	go.unistack.org/micro/v3 v3.11.41
+	golang.org/x/net v0.33.0
+	google.golang.org/grpc v1.69.2
+	google.golang.org/protobuf v1.36.1
 )

 require (
 	github.com/ash3in/uuidv8 v1.2.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
 	github.com/matoous/go-nanoid v1.5.1 // indirect
-	github.com/spf13/cast v1.7.1 // indirect
-	go.unistack.org/micro-proto/v4 v4.1.0 // indirect
-	golang.org/x/sys v0.32.0 // indirect
-	golang.org/x/text v0.24.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34 // indirect
+	go.unistack.org/micro-proto/v3 v3.4.1 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20241223144023-3abc09e42ca8 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
--- a/go.sum
+++ b/go.sum
@@ -2,10 +2,7 @@ github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7Oputl
 github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
 github.com/ash3in/uuidv8 v1.2.0 h1:2oogGdtCPwaVtyvPPGin4TfZLtOGE5F+W++E880G6SI=
 github.com/ash3in/uuidv8 v1.2.0/go.mod h1:BnU0wJBxnzdEKmVg4xckBkD+VZuecTFTUP3M0dWgyY4=
-github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
-github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
-github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
@@ -22,42 +19,36 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/matoous/go-nanoid v1.5.1 h1:aCjdvTyO9LLnTIi0fgdXhOPPvOHjpXN6Ik9DaNjIct4=
 github.com/matoous/go-nanoid v1.5.1/go.mod h1:zyD2a71IubI24efhpvkJz+ZwfwagzgSO6UNiFsZKN7U=
-github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
-github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
 github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
-github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y=
-github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
-github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
-github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
-go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
-go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY=
-go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI=
-go.opentelemetry.io/otel/metric v1.34.0 h1:+eTR3U0MyfWjRDhmFMxe2SsW64QrZ84AOhvqS7Y+PoQ=
-go.opentelemetry.io/otel/metric v1.34.0/go.mod h1:CEDrp0fy2D0MvkXE+dPV7cMi8tWZwX3dmaIhwPOaqHE=
-go.opentelemetry.io/otel/sdk v1.34.0 h1:95zS4k/2GOy069d321O8jWgYsW3MzVV+KuSPKp7Wr1A=
-go.opentelemetry.io/otel/sdk v1.34.0/go.mod h1:0e/pNiaMAqaykJGKbi+tSjWfNNHMTxoC9qANsCzbyxU=
-go.opentelemetry.io/otel/sdk/metric v1.34.0 h1:5CeK9ujjbFVL5c1PhLuStg1wxA7vQv7ce1EK0Gyvahk=
-go.opentelemetry.io/otel/sdk/metric v1.34.0/go.mod h1:jQ/r8Ze28zRKoNRdkjCZxfs6YvBTG1+YIqyFVFYec5w=
-go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC8mh/k=
-go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
-go.unistack.org/micro-proto/v4 v4.1.0 h1:qPwL2n/oqh9RE3RTTDgt28XK3QzV597VugQPaw9lKUk=
-go.unistack.org/micro-proto/v4 v4.1.0/go.mod h1:ArmK7o+uFvxSY3dbJhKBBX4Pm1rhWdLEFf3LxBrMtec=
-go.unistack.org/micro/v4 v4.1.8 h1:bbqnQ7nzTYNIYt4HhRgmh/++qzwGJvlNbrTOoXzkQiU=
-go.unistack.org/micro/v4 v4.1.8/go.mod h1:b4dr7RFlbpSfSsKCva9UuX4zLtkYQteEK+Uuac39qJE=
-golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY=
-golang.org/x/net v0.39.0/go.mod h1:X7NRbYVEA+ewNkCNyJ513WmMdQ3BineSwVtN2zD/d+E=
-golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
-golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
-golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
-golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34 h1:h6p3mQqrmT1XkHVTfzLdNz1u7IhINeZkz67/xTbOuWs=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20250428153025-10db94c68c34/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
-google.golang.org/grpc v1.72.0 h1:S7UkcVa60b5AAQTaO6ZKamFp1zMZSU0fGDK2WZLbBnM=
-google.golang.org/grpc v1.72.0/go.mod h1:wH5Aktxcg25y1I3w7H69nHfXdOG3UiadoBtjh3izSDM=
-google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
-google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
+go.opentelemetry.io/otel v1.31.0 h1:NsJcKPIW0D0H3NgzPDHmo0WW6SptzPdqg/L1zsIm2hY=
+go.opentelemetry.io/otel v1.31.0/go.mod h1:O0C14Yl9FgkjqcCZAsE053C13OaddMYr/hz6clDkEJE=
+go.opentelemetry.io/otel/metric v1.31.0 h1:FSErL0ATQAmYHUIzSezZibnyVlft1ybhy4ozRPcF2fE=
+go.opentelemetry.io/otel/metric v1.31.0/go.mod h1:C3dEloVbLuYoX41KpmAhOqNriGbA+qqH6PQ5E5mUfnY=
+go.opentelemetry.io/otel/sdk v1.31.0 h1:xLY3abVHYZ5HSfOg3l2E5LUj2Cwva5Y7yGxnSW9H5Gk=
+go.opentelemetry.io/otel/sdk v1.31.0/go.mod h1:TfRbMdhvxIIr/B2N2LQW2S5v9m3gOQ/08KsbbO5BPT0=
+go.opentelemetry.io/otel/sdk/metric v1.31.0 h1:i9hxxLJF/9kkvfHppyLL55aW7iIJz4JjxTeYusH7zMc=
+go.opentelemetry.io/otel/sdk/metric v1.31.0/go.mod h1:CRInTMVvNhUKgSAMbKyTMxqOBC0zgyxzW55lZzX43Y8=
+go.opentelemetry.io/otel/trace v1.31.0 h1:ffjsj1aRouKewfr85U2aGagJ46+MvodynlQ1HYdmJys=
+go.opentelemetry.io/otel/trace v1.31.0/go.mod h1:TXZkRk7SM2ZQLtR6eoAWQFIHPvzQ06FJAsO1tJg480A=
+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.41 h1:dP4sBLIZpMo+MWGe5bbESewK8wBzYm4Yik/67x4dEtQ=
+go.unistack.org/micro/v3 v3.11.41/go.mod h1:POGU5hstnAT9LH70m8FalyQSNi2GfIew71K75JenIZk=
+golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20241223144023-3abc09e42ca8 h1:TqExAhdPaB60Ux47Cn0oLV07rGnxZzIsaRhQaqS666A=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20241223144023-3abc09e42ca8/go.mod h1:lcTa1sDdWEIHMWlITnIczmw5w60CF9ffkb8Z+DVmmjA=
+google.golang.org/grpc v1.69.2 h1:U3S9QEtbXC0bYNvRtcoklF3xGtLViumSYxWykJS+7AU=
+google.golang.org/grpc v1.69.2/go.mod h1:vyjdE6jLBI76dgpDojsFGNaHlxdjXN9ghpnd2o7JGZ4=
+google.golang.org/protobuf v1.36.1 h1:yBPeRvTftaleIgM3PZ/WBIZ7XM/eEYAaEyCwvyjq/gk=
+google.golang.org/protobuf v1.36.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
--- a/grpc.go
+++ b/grpc.go
@@ -10,6 +10,7 @@ import (
 	"slices"
 	"sort"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -18,16 +19,18 @@ import (
 	reflectionv1pb "google.golang.org/grpc/reflection/grpc_reflection_v1"

 	// nolint: staticcheck
-
-	"go.unistack.org/micro/v4/errors"
-	"go.unistack.org/micro/v4/logger"
-	"go.unistack.org/micro/v4/metadata"
-	"go.unistack.org/micro/v4/meter"
-	"go.unistack.org/micro/v4/options"
-	"go.unistack.org/micro/v4/register"
-	"go.unistack.org/micro/v4/semconv"
-	"go.unistack.org/micro/v4/server"
-	"go.unistack.org/micro/v4/tracer"
+	"go.unistack.org/micro/v3/broker"
+	"go.unistack.org/micro/v3/codec"
+	"go.unistack.org/micro/v3/errors"
+	"go.unistack.org/micro/v3/logger"
+	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/meter"
+	"go.unistack.org/micro/v3/options"
+	"go.unistack.org/micro/v3/register"
+	"go.unistack.org/micro/v3/semconv"
+	"go.unistack.org/micro/v3/server"
+	msync "go.unistack.org/micro/v3/sync"
+	"go.unistack.org/micro/v3/tracer"
 	"golang.org/x/net/netutil"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -43,6 +46,13 @@ const (
 	DefaultContentType = "application/grpc"
 )

+/*
+type ServerReflection struct {
+	srv *grpc.Server
+	s   *serverReflectionServer
+}
+*/
+
 type streamWrapper struct {
 	ctx context.Context
 	grpc.ServerStream
@@ -59,7 +69,9 @@ type Server struct {
 	handlers       map[string]server.Handler
 	srv            *grpc.Server
 	exit           chan chan error
+	wg             *msync.WaitGroup
 	rsvc           *register.Service
+	subscribers    map[*subscriber][]broker.Subscriber
 	rpc            *rServer
 	opts           server.Options
 	unknownHandler grpc.StreamHandler
@@ -80,6 +92,7 @@ func newServer(opts ...server.Option) *Server {
 			serviceMap: make(map[string]*service),
 		},
 		handlers:    make(map[string]server.Handler),
+		subscribers: make(map[*subscriber][]broker.Subscriber),
 		exit:        make(chan chan error),
 		stateLive:   &atomic.Uint32{},
 		stateReady:  &atomic.Uint32{},
@@ -91,6 +104,22 @@ func newServer(opts ...server.Option) *Server {
 	return g
 }

+/*
+type grpcRouter struct {
+ h func(context.Context, server.Request, interface{}) error
+ m func(context.Context, server.Message) error
+}
+
+func (r grpcRouter) ProcessMessage(ctx context.Context, msg server.Message) error {
+ return r.m(ctx, msg)
+}
+
+func (r grpcRouter) ServeRequest(ctx context.Context, req server.Request, rsp server.Response) error {
+ return r.h(ctx, req, rsp)
+}
+
+*/
+
 func (g *Server) configure(opts ...server.Option) error {
 	g.Lock()
 	defer g.Unlock()
@@ -186,9 +215,8 @@ func (g *Server) handler(srv interface{}, stream grpc.ServerStream) error {
 		return status.Errorf(codes.Internal, "method does not exist in context")
 	}

-	var gmd map[string][]string
 	// get grpc metadata
-	gmd, ok = gmetadata.FromIncomingContext(ctx)
+	gmd, ok := gmetadata.FromIncomingContext(ctx)
 	if !ok {
 		gmd = gmetadata.MD{}
 	}
@@ -221,23 +249,25 @@ func (g *Server) handler(srv interface{}, stream grpc.ServerStream) error {
 		}()
 	}

-	md := metadata.Copy(gmd)
-
-	md.Set("path", fullMethod)
-	md.Set("micro-server", "grpc")
+	md := metadata.New(len(gmd))
+	for k, v := range gmd {
+		md[k] = strings.Join(v, ", ")
+	}
+	md.Set("Path", fullMethod)
+	md.Set("Micro-Server", "grpc")
 	md.Set(metadata.HeaderEndpoint, methodName)
 	md.Set(metadata.HeaderService, serviceName)

 	var td string
 	// timeout for server deadline
-	if v := md.Get("timeout"); v != nil {
+	if v, ok := md.Get("timeout"); ok {
 		md.Del("timeout")
-		td = v[0]
+		td = v
 	}
-	if v := md.Get("grpc-timeout"); v != nil {
-		md.Del("grpc-timeout")
-		td = v[0][:len(v)-1]
-		switch v[0][len(v)-1:] {
+	if v, ok := md.Get("Grpc-Timeout"); ok {
+		md.Del("Grpc-Timeout")
+		td = v[:len(v)-1]
+		switch v[len(v)-1:] {
 		case "S":
 			td += "s"
 		case "M":
@@ -256,8 +286,11 @@ func (g *Server) handler(srv interface{}, stream grpc.ServerStream) error {
 	// get content type
 	ct := DefaultContentType

-	if ctype := md.Get("content-type"); ctype != nil {
-		ct = ctype[0]
+	if ctype, ok := md.Get("content-type"); ok {
+		ct = ctype
+	} else if ctype, ok := md.Get("x-content-type"); ok {
+		ct = ctype
+		md.Del("x-content-type")
 	}

 	// create new context
@@ -290,7 +323,7 @@ func (g *Server) handler(srv interface{}, stream grpc.ServerStream) error {

 	// get peer from context
 	if p, ok := peer.FromContext(ctx); ok {
-		md.Set("remote", p.Addr.String())
+		md.Set("Remote", p.Addr.String())
 		ctx = peer.NewContext(ctx, p)
 	}

@@ -398,7 +431,7 @@ func (g *Server) processRequest(ctx context.Context, stream grpc.ServerStream, s
 	// execute the handler
 	appErr := fn(ctx, r, replyv.Interface())
 	if outmd, ok := metadata.FromOutgoingContext(ctx); ok {
-		if err = stream.SendHeader(gmetadata.MD(outmd.Copy())); err != nil {
+		if err = stream.SendHeader(gmetadata.New(outmd)); err != nil {
 			return err
 		}
 	}
@@ -482,7 +515,7 @@ func (g *Server) processStream(ctx context.Context, stream grpc.ServerStream, se

 	appErr := fn(ctx, r, ss)
 	if outmd, ok := metadata.FromOutgoingContext(ctx); ok {
-		if err := stream.SendHeader(gmetadata.MD(outmd.Copy())); err != nil {
+		if err := stream.SendHeader(gmetadata.New(outmd)); err != nil {
 			return err
 		}
 	}
@@ -518,6 +551,21 @@ func (g *Server) processStream(ctx context.Context, stream grpc.ServerStream, se
 	return status.New(statusCode, statusDesc).Err()
 }

+func (g *Server) newCodec(ct string) (codec.Codec, error) {
+	g.RLock()
+	defer g.RUnlock()
+
+	if idx := strings.IndexRune(ct, ';'); idx >= 0 {
+		ct = ct[:idx]
+	}
+
+	if c, ok := g.opts.Codecs[ct]; ok {
+		return c, nil
+	}
+
+	return nil, codec.ErrUnknownContentType
+}
+
 func (g *Server) Options() server.Options {
 	g.RLock()
 	opts := g.opts
@@ -543,6 +591,34 @@ func (g *Server) Handle(h server.Handler) error {
 	return nil
 }

+func (g *Server) NewSubscriber(topic string, sb interface{}, opts ...server.SubscriberOption) server.Subscriber {
+	return newSubscriber(topic, sb, opts...)
+}
+
+func (g *Server) Subscribe(sb server.Subscriber) error {
+	sub, ok := sb.(*subscriber)
+	if !ok {
+		return fmt.Errorf("invalid subscriber: expected *subscriber")
+	}
+	if len(sub.handlers) == 0 {
+		return fmt.Errorf("invalid subscriber: no handler functions")
+	}
+
+	if err := server.ValidateSubscriber(sb); err != nil {
+		return err
+	}
+
+	g.Lock()
+	if _, ok = g.subscribers[sub]; ok {
+		g.Unlock()
+		return fmt.Errorf("subscriber %v already exists", sub)
+	}
+
+	g.subscribers[sub] = nil
+	g.Unlock()
+	return nil
+}
+
 func (g *Server) Register() error {
 	g.RLock()
 	rsvc := g.rsvc
@@ -572,6 +648,15 @@ func (g *Server) Register() error {

 	sort.Strings(handlerList)

+	subscriberList := make([]*subscriber, 0, len(g.subscribers))
+	for e := range g.subscribers {
+		// Only advertise non internal subscribers
+		subscriberList = append(subscriberList, e)
+	}
+	sort.Slice(subscriberList, func(i, j int) bool {
+		return subscriberList[i].topic > subscriberList[j].topic
+	})
+
 	g.RUnlock()

 	g.RLock()
@@ -633,6 +718,26 @@ func (g *Server) Deregister() error {

 	g.registered = false

+	wg := sync.WaitGroup{}
+	for sb, subs := range g.subscribers {
+		for _, sub := range subs {
+			wg.Add(1)
+			go func(s broker.Subscriber) {
+				defer wg.Done()
+				if config.Logger.V(logger.InfoLevel) {
+					config.Logger.Info(config.Context, "Unsubscribing from topic: "+s.Topic())
+				}
+				if err := s.Unsubscribe(g.opts.Context); err != nil {
+					if config.Logger.V(logger.ErrorLevel) {
+						config.Logger.Error(config.Context, "Unsubscribing from topic: "+s.Topic(), err)
+					}
+				}
+			}(sub)
+		}
+		g.subscribers[sb] = nil
+	}
+	wg.Wait()
+
 	g.Unlock()
 	return nil
 }
@@ -680,6 +785,21 @@ func (g *Server) Start() error {
 	}
 	g.Unlock()

+	// only connect if we're subscribed
+	if len(g.subscribers) > 0 {
+		// connect to the broker
+		if err = config.Broker.Connect(config.Context); err != nil {
+			if config.Logger.V(logger.ErrorLevel) {
+				config.Logger.Error(config.Context, fmt.Sprintf("broker [%s] connect error", config.Broker.String()), err)
+			}
+			return err
+		}
+
+		if config.Logger.V(logger.InfoLevel) {
+			config.Logger.Info(config.Context, fmt.Sprintf("broker [%s] Connected to %s", config.Broker.String(), config.Broker.Address()))
+		}
+	}
+
 	// use RegisterCheck func before register
 	// nolint: nestif
 	if err = g.opts.RegisterCheck(config.Context); err != nil {
@@ -695,6 +815,10 @@ func (g *Server) Start() error {
 		}
 	}

+	if err = g.subscribe(); err != nil {
+		return err
+	}
+
 	// micro: go ts.Accept(s.accept)
 	go func() {
 		if err = g.srv.Serve(ts); err != nil {
--- a/handler.go
+++ b/handler.go
@@ -3,7 +3,7 @@ package grpc
 import (
 	"reflect"

-	"go.unistack.org/micro/v4/server"
+	"go.unistack.org/micro/v3/server"
 )

 type rpcHandler struct {
--- a/options.go
+++ b/options.go
@@ -3,7 +3,7 @@ package grpc
 import (
 	"context"

-	"go.unistack.org/micro/v4/server"
+	"go.unistack.org/micro/v3/server"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/encoding"
 )
--- a/reflection_test.go
+++ b/reflection_test.go
@@ -3,8 +3,9 @@ package grpc
 import (
 	"fmt"
 	"testing"
-	
-	"go.unistack.org/micro/v4/server"
+
+	_ "go.unistack.org/micro-server-grpc/v3/proto"
+	"go.unistack.org/micro/v3/server"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/reflect/protoreflect"
 	"google.golang.org/protobuf/reflect/protoregistry"
--- a/request.go
+++ b/request.go
@@ -1,9 +1,9 @@
 package grpc

 import (
-	"go.unistack.org/micro/v4/codec"
-	"go.unistack.org/micro/v4/metadata"
-	"go.unistack.org/micro/v4/server"
+	"go.unistack.org/micro/v3/codec"
+	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/server"
 )

 var _ server.Request = &rpcRequest{}
--- a/response.go
+++ b/response.go
@@ -1,9 +1,9 @@
 package grpc

 import (
-	"go.unistack.org/micro/v4/codec"
-	"go.unistack.org/micro/v4/metadata"
-	"go.unistack.org/micro/v4/server"
+	"go.unistack.org/micro/v3/codec"
+	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/server"
 )

 var _ server.Response = &rpcResponse{}
--- a/server.go
+++ b/server.go
@@ -14,7 +14,7 @@ import (
 	"unicode"
 	"unicode/utf8"

-	"go.unistack.org/micro/v4/server"
+	"go.unistack.org/micro/v3/server"
 )

 // Precompute the reflect type for error. Can't use error directly
@@ -47,8 +47,8 @@ type rServer struct {

 // Is this an exported - upper case - name?
 func isExported(name string) bool {
-	r, _ := utf8.DecodeRuneInString(name)
-	return unicode.IsUpper(r)
+	rune, _ := utf8.DecodeRuneInString(name)
+	return unicode.IsUpper(rune)
 }

 // Is this type exported or a builtin?
@@ -123,43 +123,43 @@ func prepareEndpoint(method reflect.Method) (*methodType, error) {
 	return &methodType{method: method, ArgType: argType, ReplyType: replyType, ContextType: contextType, stream: stream}, nil
 }

-func (s *rServer) register(rcvr interface{}) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if s.serviceMap == nil {
-		s.serviceMap = make(map[string]*service)
+func (server *rServer) register(rcvr interface{}) error {
+	server.mu.Lock()
+	defer server.mu.Unlock()
+	if server.serviceMap == nil {
+		server.serviceMap = make(map[string]*service)
 	}
-	srv := &service{}
-	srv.typ = reflect.TypeOf(rcvr)
-	srv.rcvr = reflect.ValueOf(rcvr)
-	sname := reflect.Indirect(srv.rcvr).Type().Name()
+	s := &service{}
+	s.typ = reflect.TypeOf(rcvr)
+	s.rcvr = reflect.ValueOf(rcvr)
+	sname := reflect.Indirect(s.rcvr).Type().Name()
 	if sname == "" {
-		return fmt.Errorf("rpc: no service name for type %v", srv.typ.String())
+		return fmt.Errorf("rpc: no service name for type %v", s.typ.String())
 	}
 	if !isExported(sname) {
 		return fmt.Errorf("rpc Register: type %s is not exported", sname)
 	}
-	if _, present := s.serviceMap[sname]; present {
+	if _, present := server.serviceMap[sname]; present {
 		return fmt.Errorf("rpc: service already defined: %s", sname)
 	}
-	srv.name = sname
-	srv.method = make(map[string]*methodType)
+	s.name = sname
+	s.method = make(map[string]*methodType)

 	// Install the methods
-	for m := 0; m < srv.typ.NumMethod(); m++ {
-		method := srv.typ.Method(m)
+	for m := 0; m < s.typ.NumMethod(); m++ {
+		method := s.typ.Method(m)
 		mt, err := prepareEndpoint(method)
 		if mt != nil && err == nil {
-			srv.method[method.Name] = mt
+			s.method[method.Name] = mt
 		} else if err != nil {
 			return err
 		}
 	}

-	if len(srv.method) == 0 {
+	if len(s.method) == 0 {
 		return fmt.Errorf("rpc Register: type %s has no exported methods of suitable type", sname)
 	}
-	s.serviceMap[srv.name] = srv
+	server.serviceMap[s.name] = s
 	return nil
 }

--- a/stream.go
+++ b/stream.go
@@ -3,7 +3,7 @@ package grpc
 import (
 	"context"

-	"go.unistack.org/micro/v4/server"
+	"go.unistack.org/micro/v3/server"
 	"google.golang.org/grpc"
 )

--- a/subscriber.go
+++ b/subscriber.go
@@ -0,0 +1,259 @@
+package grpc
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"strings"
+
+	"go.unistack.org/micro/v3/broker"
+	"go.unistack.org/micro/v3/codec"
+	"go.unistack.org/micro/v3/logger"
+	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/options"
+	"go.unistack.org/micro/v3/server"
+)
+
+var _ server.Message = &rpcMessage{}
+
+type rpcMessage struct {
+	payload     interface{}
+	codec       codec.Codec
+	header      metadata.Metadata
+	topic       string
+	contentType string
+}
+
+func (r *rpcMessage) ContentType() string {
+	return r.contentType
+}
+
+func (r *rpcMessage) Topic() string {
+	return r.topic
+}
+
+func (r *rpcMessage) Body() interface{} {
+	return r.payload
+}
+
+func (r *rpcMessage) Header() metadata.Metadata {
+	return r.header
+}
+
+func (r *rpcMessage) Codec() codec.Codec {
+	return r.codec
+}
+
+type handler struct {
+	reqType reflect.Type
+	ctxType reflect.Type
+	method  reflect.Value
+}
+
+type subscriber struct {
+	topic      string
+	rcvr       reflect.Value
+	typ        reflect.Type
+	subscriber interface{}
+	handlers   []*handler
+	opts       server.SubscriberOptions
+}
+
+func newSubscriber(topic string, sub interface{}, opts ...server.SubscriberOption) server.Subscriber {
+	options := server.NewSubscriberOptions(opts...)
+
+	var handlers []*handler
+
+	if typ := reflect.TypeOf(sub); typ.Kind() == reflect.Func {
+		h := &handler{
+			method: reflect.ValueOf(sub),
+		}
+
+		switch typ.NumIn() {
+		case 1:
+			h.reqType = typ.In(0)
+		case 2:
+			h.ctxType = typ.In(0)
+			h.reqType = typ.In(1)
+		}
+
+		handlers = append(handlers, h)
+
+	} else {
+		for m := 0; m < typ.NumMethod(); m++ {
+			method := typ.Method(m)
+			h := &handler{
+				method: method.Func,
+			}
+
+			switch method.Type.NumIn() {
+			case 2:
+				h.reqType = method.Type.In(1)
+			case 3:
+				h.ctxType = method.Type.In(1)
+				h.reqType = method.Type.In(2)
+			}
+
+			handlers = append(handlers, h)
+
+		}
+	}
+
+	return &subscriber{
+		rcvr:       reflect.ValueOf(sub),
+		typ:        reflect.TypeOf(sub),
+		topic:      topic,
+		subscriber: sub,
+		handlers:   handlers,
+		opts:       options,
+	}
+}
+
+func (g *Server) createSubHandler(sb *subscriber, opts server.Options) broker.Handler {
+	return func(p broker.Event) (err error) {
+		msg := p.Message()
+		// if we don't have headers, create empty map
+		if msg.Header == nil {
+			msg.Header = make(map[string]string)
+		}
+
+		ct := msg.Header["Content-Type"]
+		if len(ct) == 0 {
+			msg.Header["Content-Type"] = DefaultContentType
+			ct = DefaultContentType
+		}
+		cf, err := g.newCodec(ct)
+		if err != nil {
+			return err
+		}
+
+		hdr := make(map[string]string, len(msg.Header))
+		for k, v := range msg.Header {
+			hdr[k] = v
+		}
+
+		ctx := metadata.NewIncomingContext(sb.opts.Context, hdr)
+		ctx = metadata.NewOutgoingContext(ctx, metadata.New(0))
+
+		results := make(chan error, len(sb.handlers))
+
+		for i := 0; i < len(sb.handlers); i++ {
+			handler := sb.handlers[i]
+
+			var isVal bool
+			var req reflect.Value
+
+			if handler.reqType.Kind() == reflect.Ptr {
+				req = reflect.New(handler.reqType.Elem())
+			} else {
+				req = reflect.New(handler.reqType)
+				isVal = true
+			}
+			if isVal {
+				req = req.Elem()
+			}
+
+			if err = cf.Unmarshal(msg.Body, req.Interface()); err != nil {
+				return err
+			}
+
+			fn := func(ctx context.Context, msg server.Message) error {
+				var vals []reflect.Value
+				if sb.typ.Kind() != reflect.Func {
+					vals = append(vals, sb.rcvr)
+				}
+				if handler.ctxType != nil {
+					vals = append(vals, reflect.ValueOf(ctx))
+				}
+
+				vals = append(vals, reflect.ValueOf(msg.Body()))
+
+				returnValues := handler.method.Call(vals)
+				if rerr := returnValues[0].Interface(); rerr != nil {
+					return rerr.(error)
+				}
+				return nil
+			}
+
+			opts.Hooks.EachPrev(func(hook options.Hook) {
+				if h, ok := hook.(server.HookSubHandler); ok {
+					fn = h(fn)
+				}
+			})
+
+			if g.wg != nil {
+				g.wg.Add(1)
+			}
+			go func() {
+				if g.wg != nil {
+					defer g.wg.Done()
+				}
+				cerr := fn(ctx, &rpcMessage{
+					topic:       sb.topic,
+					contentType: ct,
+					payload:     req.Interface(),
+					header:      msg.Header,
+				})
+				results <- cerr
+			}()
+		}
+		var errors []string
+		for i := 0; i < len(sb.handlers); i++ {
+			if rerr := <-results; rerr != nil {
+				errors = append(errors, rerr.Error())
+			}
+		}
+		if len(errors) > 0 {
+			err = fmt.Errorf("subscriber error: %s", strings.Join(errors, "\n"))
+		}
+
+		return err
+	}
+}
+
+func (s *subscriber) Topic() string {
+	return s.topic
+}
+
+func (s *subscriber) Subscriber() interface{} {
+	return s.subscriber
+}
+
+func (s *subscriber) Options() server.SubscriberOptions {
+	return s.opts
+}
+
+func (g *Server) subscribe() error {
+	config := g.opts
+	subCtx := config.Context
+
+	for sb := range g.subscribers {
+
+		if cx := sb.Options().Context; cx != nil {
+			subCtx = cx
+		}
+
+		opts := []broker.SubscribeOption{
+			broker.SubscribeContext(subCtx),
+			broker.SubscribeAutoAck(sb.Options().AutoAck),
+			broker.SubscribeBodyOnly(sb.Options().BodyOnly),
+		}
+
+		if queue := sb.Options().Queue; len(queue) > 0 {
+			opts = append(opts, broker.SubscribeGroup(queue))
+		}
+
+		if config.Logger.V(logger.InfoLevel) {
+			config.Logger.Info(config.Context, "subscribing to topic: "+sb.Topic())
+		}
+
+		sub, err := config.Broker.Subscribe(subCtx, sb.Topic(), g.createSubHandler(sb, config), opts...)
+		if err != nil {
+			return err
+		}
+
+		g.subscribers[sb] = []broker.Subscriber{sub}
+	}
+
+	return nil
+}
Author	SHA1	Message	Date
pugnack	fc50af50bb	add missing outgoing key to ctx for subscribers (#279 ) All checks were successful coverage / build (push) Successful in 1m5s Details test / test (push) Successful in 4m52s Details	2025-06-11 01:22:10 +03:00
pugnack	19a8d7f7b1	update ci (#276 )	2025-05-05 19:24:46 +03:00
pugnack	8b3b2700d6	disable coverage job (#274 )	2025-05-03 10:39:55 +03:00
vtolstov	6cdb4ff377	Apply Code Coverage Badge	2025-05-02 16:14:36 +00:00
pugnack	b104e25898	update ci (#266 ) All checks were successful coverage / build (push) Successful in 3m4s Details test / test (push) Successful in 5m50s Details	2025-05-02 19:10:38 +03:00
Vasiliy Tolstov	8102d709f4	Rename gitea to GitHub v3 (#259 ) * rename .gitea to .github * attempt to fix lint/test job * attempt to fix lint/test job --------- Co-authored-by: pugnack <pugnack@gmail.com>	2025-04-29 18:12:45 +03:00