You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

41 lines
1.4 KiB

  1. package pub_client
  2. import (
  3. "fmt"
  4. "github.com/seaweedfs/seaweedfs/weed/mq/balancer"
  5. "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
  6. "github.com/seaweedfs/seaweedfs/weed/util"
  7. )
  8. func (p *TopicPublisher) Publish(key, value []byte) error {
  9. hashKey := util.HashToInt32(key) % balancer.MaxPartitionCount
  10. if hashKey < 0 {
  11. hashKey = -hashKey
  12. }
  13. publishClient, found := p.partition2Broker.Floor(hashKey, hashKey)
  14. if !found {
  15. return fmt.Errorf("no broker found for key %d", hashKey)
  16. }
  17. p.Lock()
  18. defer p.Unlock()
  19. // dead lock here
  20. //google.golang.org/grpc/internal/transport.(*writeQuota).get(flowcontrol.go:59)
  21. //google.golang.org/grpc/internal/transport.(*http2Client).Write(http2_client.go:1047)
  22. //google.golang.org/grpc.(*csAttempt).sendMsg(stream.go:1040)
  23. //google.golang.org/grpc.(*clientStream).SendMsg.func2(stream.go:892)
  24. //google.golang.org/grpc.(*clientStream).withRetry(stream.go:752)
  25. //google.golang.org/grpc.(*clientStream).SendMsg(stream.go:894)
  26. //github.com/seaweedfs/seaweedfs/weed/pb/mq_pb.(*seaweedMessagingPublishClient).Send(mq_grpc.pb.go:141)
  27. //github.com/seaweedfs/seaweedfs/weed/mq/client/pub_client.(*TopicPublisher).Publish(publish.go:19)
  28. if err := publishClient.Send(&mq_pb.PublishRequest{
  29. Message: &mq_pb.PublishRequest_Data{
  30. Data: &mq_pb.DataMessage{
  31. Key: key,
  32. Value: value,
  33. },
  34. },
  35. }); err != nil {
  36. return fmt.Errorf("send publish request: %v", err)
  37. }
  38. return nil
  39. }