| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| package internal |
| |
| import ( |
| "crypto/tls" |
| "crypto/x509" |
| "errors" |
| "fmt" |
| "net" |
| "net/url" |
| "os" |
| "sync" |
| "sync/atomic" |
| "time" |
| |
| "github.com/apache/pulsar-client-go/pulsar/auth" |
| |
| "google.golang.org/protobuf/proto" |
| |
| pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto" |
| "github.com/apache/pulsar-client-go/pulsar/log" |
| |
| ua "go.uber.org/atomic" |
| ) |
| |
| const ( |
| PulsarProtocolVersion = int32(pb.ProtocolVersion_v18) |
| ) |
| |
| type TLSOptions struct { |
| KeyFile string |
| CertFile string |
| TrustCertsFilePath string |
| AllowInsecureConnection bool |
| ValidateHostname bool |
| ServerName string |
| CipherSuites []uint16 |
| MinVersion uint16 |
| MaxVersion uint16 |
| } |
| |
| var ( |
| errConnectionClosed = errors.New("connection closed") |
| errUnableRegisterListener = errors.New("unable register listener when con closed") |
| errUnableAddConsumeHandler = errors.New("unable add consumer handler when con closed") |
| ) |
| |
| // ConnectionListener is a user of a connection (eg. a producer or |
| // a consumer) that can register itself to get notified |
| // when the connection is closed. |
| type ConnectionListener interface { |
| // ReceivedSendReceipt receive and process the return value of the send command. |
| ReceivedSendReceipt(response *pb.CommandSendReceipt) |
| |
| // ConnectionClosed close the TCP connection. |
| ConnectionClosed(closeProducer *pb.CommandCloseProducer) |
| } |
| |
| // Connection is a interface of client cnx. |
| type Connection interface { |
| SendRequest(requestID uint64, req *pb.BaseCommand, callback func(*pb.BaseCommand, error)) |
| SendRequestNoWait(req *pb.BaseCommand) error |
| WriteData(data Buffer) |
| RegisterListener(id uint64, listener ConnectionListener) error |
| UnregisterListener(id uint64) |
| AddConsumeHandler(id uint64, handler ConsumerHandler) error |
| DeleteConsumeHandler(id uint64) |
| ID() string |
| GetMaxMessageSize() int32 |
| Close() |
| IsProxied() bool |
| } |
| |
| type ConsumerHandler interface { |
| MessageReceived(response *pb.CommandMessage, headersAndPayload Buffer) error |
| |
| ActiveConsumerChanged(isActive bool) |
| |
| // ConnectionClosed close the TCP connection. |
| ConnectionClosed(closeConsumer *pb.CommandCloseConsumer) |
| } |
| |
| type connectionState int32 |
| |
| const ( |
| connectionInit = iota |
| connectionReady |
| connectionClosing |
| connectionClosed |
| ) |
| |
| func (s connectionState) String() string { |
| switch s { |
| case connectionInit: |
| return "Initializing" |
| case connectionReady: |
| return "Ready" |
| case connectionClosing: |
| return "Closing" |
| case connectionClosed: |
| return "Closed" |
| default: |
| return "Unknown" |
| } |
| } |
| |
| type request struct { |
| id *uint64 |
| cmd *pb.BaseCommand |
| callback func(command *pb.BaseCommand, err error) |
| } |
| |
| type incomingCmd struct { |
| cmd *pb.BaseCommand |
| headersAndPayload Buffer |
| } |
| |
| type connection struct { |
| sync.Mutex |
| cond *sync.Cond |
| started int32 |
| state ua.Int32 |
| connectionTimeout time.Duration |
| closeOnce sync.Once |
| |
| logicalAddr *url.URL |
| physicalAddr *url.URL |
| cnx net.Conn |
| |
| writeBufferLock sync.Mutex |
| writeBuffer Buffer |
| reader *connectionReader |
| |
| lastDataReceivedLock sync.Mutex |
| lastDataReceivedTime time.Time |
| |
| log log.Logger |
| |
| incomingRequestsWG sync.WaitGroup |
| incomingRequestsCh chan *request |
| incomingCmdCh chan *incomingCmd |
| closeCh chan interface{} |
| writeRequestsCh chan Buffer |
| |
| pendingLock sync.Mutex |
| pendingReqs map[uint64]*request |
| |
| listenersLock sync.RWMutex |
| listeners map[uint64]ConnectionListener |
| |
| consumerHandlersLock sync.RWMutex |
| consumerHandlers map[uint64]ConsumerHandler |
| |
| tlsOptions *TLSOptions |
| auth auth.Provider |
| |
| maxMessageSize int32 |
| metrics *Metrics |
| |
| keepAliveInterval time.Duration |
| |
| lastActive time.Time |
| } |
| |
| // connectionOptions defines configurations for creating connection. |
| type connectionOptions struct { |
| logicalAddr *url.URL |
| physicalAddr *url.URL |
| tls *TLSOptions |
| connectionTimeout time.Duration |
| auth auth.Provider |
| logger log.Logger |
| metrics *Metrics |
| keepAliveInterval time.Duration |
| } |
| |
| func newConnection(opts connectionOptions) *connection { |
| cnx := &connection{ |
| connectionTimeout: opts.connectionTimeout, |
| keepAliveInterval: opts.keepAliveInterval, |
| logicalAddr: opts.logicalAddr, |
| physicalAddr: opts.physicalAddr, |
| writeBuffer: NewBuffer(4096), |
| log: opts.logger.SubLogger(log.Fields{"remote_addr": opts.physicalAddr}), |
| pendingReqs: make(map[uint64]*request), |
| lastDataReceivedTime: time.Now(), |
| tlsOptions: opts.tls, |
| auth: opts.auth, |
| |
| closeCh: make(chan interface{}), |
| incomingRequestsCh: make(chan *request, 10), |
| incomingCmdCh: make(chan *incomingCmd, 10), |
| |
| // This channel is used to pass data from producers to the connection |
| // go routine. It can become contended or blocking if we have multiple |
| // partition produces writing on a single connection. In general it's |
| // good to keep this above the number of partition producers assigned |
| // to a single connection. |
| writeRequestsCh: make(chan Buffer, 256), |
| listeners: make(map[uint64]ConnectionListener), |
| consumerHandlers: make(map[uint64]ConsumerHandler), |
| metrics: opts.metrics, |
| } |
| cnx.setState(connectionInit) |
| cnx.reader = newConnectionReader(cnx) |
| cnx.cond = sync.NewCond(cnx) |
| return cnx |
| } |
| |
| func (c *connection) start() { |
| if !atomic.CompareAndSwapInt32(&c.started, 0, 1) { |
| c.log.Warnf("connection has already started") |
| return |
| } |
| |
| // Each connection gets its own goroutine that will |
| go func() { |
| if c.connect() { |
| if c.doHandshake() { |
| c.metrics.ConnectionsOpened.Inc() |
| c.run() |
| } else { |
| c.metrics.ConnectionsHandshakeErrors.Inc() |
| c.Close() |
| } |
| } else { |
| c.metrics.ConnectionsEstablishmentErrors.Inc() |
| c.Close() |
| } |
| }() |
| } |
| |
| func (c *connection) connect() bool { |
| c.log.Info("Connecting to broker") |
| |
| var ( |
| err error |
| cnx net.Conn |
| tlsConfig *tls.Config |
| ) |
| |
| if c.tlsOptions == nil { |
| // Clear text connection |
| if c.connectionTimeout.Nanoseconds() > 0 { |
| cnx, err = net.DialTimeout("tcp", c.physicalAddr.Host, c.connectionTimeout) |
| } else { |
| cnx, err = net.Dial("tcp", c.physicalAddr.Host) |
| } |
| } else { |
| // TLS connection |
| tlsConfig, err = c.getTLSConfig() |
| if err != nil { |
| c.log.WithError(err).Warn("Failed to configure TLS ") |
| return false |
| } |
| |
| // time.Duration is initialized to 0 by default, net.Dialer's default timeout is no timeout |
| // therefore if c.connectionTimeout is 0, it means no timeout |
| d := &net.Dialer{Timeout: c.connectionTimeout} |
| cnx, err = tls.DialWithDialer(d, "tcp", c.physicalAddr.Host, tlsConfig) |
| } |
| |
| if err != nil { |
| c.log.WithError(err).Warn("Failed to connect to broker.") |
| c.Close() |
| return false |
| } |
| |
| c.Lock() |
| c.cnx = cnx |
| c.log = c.log.SubLogger(log.Fields{"local_addr": c.cnx.LocalAddr()}) |
| c.log.Info("TCP connection established") |
| c.Unlock() |
| |
| return true |
| } |
| |
| func (c *connection) doHandshake() bool { |
| // Send 'Connect' command to initiate handshake |
| authData, err := c.auth.GetData() |
| if err != nil { |
| c.log.WithError(err).Warn("Failed to load auth credentials") |
| return false |
| } |
| |
| // During the initial handshake, the internal keep alive is not |
| // active yet, so we need to timeout write and read requests |
| c.cnx.SetDeadline(time.Now().Add(c.keepAliveInterval)) |
| cmdConnect := &pb.CommandConnect{ |
| ProtocolVersion: proto.Int32(PulsarProtocolVersion), |
| ClientVersion: proto.String(ClientVersionString), |
| AuthMethodName: proto.String(c.auth.Name()), |
| AuthData: authData, |
| FeatureFlags: &pb.FeatureFlags{ |
| SupportsAuthRefresh: proto.Bool(true), |
| SupportsBrokerEntryMetadata: proto.Bool(true), |
| }, |
| } |
| |
| if c.IsProxied() { |
| cmdConnect.ProxyToBrokerUrl = proto.String(c.logicalAddr.Host) |
| } |
| c.writeCommand(baseCommand(pb.BaseCommand_CONNECT, cmdConnect)) |
| cmd, _, err := c.reader.readSingleCommand() |
| if err != nil { |
| c.log.WithError(err).Warn("Failed to perform initial handshake") |
| return false |
| } |
| |
| // Reset the deadline so that we don't use read timeouts |
| c.cnx.SetDeadline(time.Time{}) |
| |
| if cmd.Connected == nil { |
| c.log.Warnf("Failed to establish connection with broker: '%s'", |
| cmd.Error.GetMessage()) |
| return false |
| } |
| if cmd.Connected.MaxMessageSize != nil && *cmd.Connected.MaxMessageSize > 0 { |
| c.log.Debug("Got MaxMessageSize from handshake response:", *cmd.Connected.MaxMessageSize) |
| c.maxMessageSize = *cmd.Connected.MaxMessageSize |
| } else { |
| c.log.Debug("No MaxMessageSize from handshake response, use default: ", MaxMessageSize) |
| c.maxMessageSize = MaxMessageSize |
| } |
| c.log.Info("Connection is ready") |
| c.setLastDataReceived(time.Now()) |
| c.changeState(connectionReady) |
| return true |
| } |
| |
| func (c *connection) IsProxied() bool { |
| return c.logicalAddr.Host != c.physicalAddr.Host |
| } |
| |
| func (c *connection) waitUntilReady() error { |
| // If we are going to call cond.Wait() at all, then we must call it _before_ we call cond.Broadcast(). |
| // The lock is held here to prevent changeState() from calling cond.Broadcast() in the time between |
| // the state check and call to cond.Wait(). |
| c.Lock() |
| defer c.Unlock() |
| |
| for c.getState() != connectionReady { |
| c.log.Debugf("Wait until connection is ready state=%s", c.getState().String()) |
| if c.getState() == connectionClosed { |
| return errors.New("connection error") |
| } |
| // wait for a new connection state change |
| c.cond.Wait() |
| } |
| |
| return nil |
| } |
| |
| func (c *connection) failLeftRequestsWhenClose() { |
| // wait for outstanding incoming requests to complete before draining |
| // and closing the channel |
| c.incomingRequestsWG.Wait() |
| |
| ch := c.incomingRequestsCh |
| go func() { |
| // send a nil message to drain instead of |
| // closing the channel and causing a potential panic |
| // |
| // if other requests come in after the nil message |
| // then the RPC client will time out |
| ch <- nil |
| }() |
| for req := range ch { |
| if nil == req { |
| break // we have drained the requests |
| } |
| c.internalSendRequest(req) |
| } |
| } |
| |
| func (c *connection) run() { |
| pingSendTicker := time.NewTicker(c.keepAliveInterval) |
| pingCheckTicker := time.NewTicker(c.keepAliveInterval) |
| |
| defer func() { |
| // stop tickers |
| pingSendTicker.Stop() |
| pingCheckTicker.Stop() |
| |
| // all the accesses to the pendingReqs should be happened in this run loop thread, |
| // including the final cleanup, to avoid the issue |
| // https://github.com/apache/pulsar-client-go/issues/239 |
| c.failPendingRequests(errConnectionClosed) |
| c.Close() |
| }() |
| |
| // All reads come from the reader goroutine |
| go c.reader.readFromConnection() |
| go c.runPingCheck(pingCheckTicker) |
| |
| c.log.Debugf("Connection run starting with request capacity=%d queued=%d", |
| cap(c.incomingRequestsCh), len(c.incomingRequestsCh)) |
| |
| go func() { |
| for { |
| select { |
| case <-c.closeCh: |
| c.failLeftRequestsWhenClose() |
| return |
| |
| case req := <-c.incomingRequestsCh: |
| if req == nil { |
| return // TODO: this never gonna be happen |
| } |
| c.internalSendRequest(req) |
| } |
| } |
| }() |
| |
| for { |
| select { |
| case <-c.closeCh: |
| return |
| |
| case cmd := <-c.incomingCmdCh: |
| c.internalReceivedCommand(cmd.cmd, cmd.headersAndPayload) |
| case data := <-c.writeRequestsCh: |
| if data == nil { |
| return |
| } |
| c.internalWriteData(data) |
| |
| case <-pingSendTicker.C: |
| c.sendPing() |
| } |
| } |
| } |
| |
| func (c *connection) runPingCheck(pingCheckTicker *time.Ticker) { |
| for { |
| select { |
| case <-c.closeCh: |
| return |
| case <-pingCheckTicker.C: |
| if c.lastDataReceived().Add(2 * c.keepAliveInterval).Before(time.Now()) { |
| // We have not received a response to the previous Ping request, the |
| // connection to broker is stale |
| c.log.Warn("Detected stale connection to broker") |
| c.Close() |
| return |
| } |
| } |
| } |
| } |
| |
| func (c *connection) WriteData(data Buffer) { |
| select { |
| case c.writeRequestsCh <- data: |
| // Channel is not full |
| return |
| |
| default: |
| // Channel full, fallback to probe if connection is closed |
| } |
| |
| for { |
| select { |
| case c.writeRequestsCh <- data: |
| // Successfully wrote on the channel |
| return |
| |
| case <-time.After(100 * time.Millisecond): |
| // The channel is either: |
| // 1. blocked, in which case we need to wait until we have space |
| // 2. the connection is already closed, then we need to bail out |
| c.log.Debug("Couldn't write on connection channel immediately") |
| state := c.getState() |
| if state != connectionReady { |
| c.log.Debug("Connection was already closed") |
| return |
| } |
| } |
| } |
| |
| } |
| |
| func (c *connection) internalWriteData(data Buffer) { |
| c.log.Debug("Write data: ", data.ReadableBytes()) |
| if _, err := c.cnx.Write(data.ReadableSlice()); err != nil { |
| c.log.WithError(err).Warn("Failed to write on connection") |
| c.Close() |
| } |
| } |
| |
| func (c *connection) writeCommand(cmd *pb.BaseCommand) { |
| // Wire format |
| // [FRAME_SIZE] [CMD_SIZE][CMD] |
| cmdSize := uint32(proto.Size(cmd)) |
| frameSize := cmdSize + 4 |
| |
| c.writeBufferLock.Lock() |
| defer c.writeBufferLock.Unlock() |
| |
| c.writeBuffer.Clear() |
| c.writeBuffer.WriteUint32(frameSize) |
| |
| c.writeBuffer.WriteUint32(cmdSize) |
| c.writeBuffer.ResizeIfNeeded(cmdSize) |
| err := MarshalToSizedBuffer(cmd, c.writeBuffer.WritableSlice()[:cmdSize]) |
| if err != nil { |
| c.log.WithError(err).Error("Protobuf serialization error") |
| panic("Protobuf serialization error") |
| } |
| |
| c.writeBuffer.WrittenBytes(cmdSize) |
| c.internalWriteData(c.writeBuffer) |
| } |
| |
| func (c *connection) receivedCommand(cmd *pb.BaseCommand, headersAndPayload Buffer) { |
| c.incomingCmdCh <- &incomingCmd{cmd, headersAndPayload} |
| } |
| |
| func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayload Buffer) { |
| c.log.Debugf("Received command: %s -- payload: %v", cmd, headersAndPayload) |
| c.setLastDataReceived(time.Now()) |
| |
| switch *cmd.Type { |
| case pb.BaseCommand_SUCCESS: |
| c.handleResponse(cmd.Success.GetRequestId(), cmd) |
| |
| case pb.BaseCommand_PRODUCER_SUCCESS: |
| if !cmd.ProducerSuccess.GetProducerReady() { |
| request, ok := c.findPendingRequest(cmd.ProducerSuccess.GetRequestId()) |
| if ok { |
| request.callback(cmd, nil) |
| } |
| } else { |
| c.handleResponse(cmd.ProducerSuccess.GetRequestId(), cmd) |
| } |
| case pb.BaseCommand_PARTITIONED_METADATA_RESPONSE: |
| c.checkServerError(cmd.PartitionMetadataResponse.Error) |
| c.handleResponse(cmd.PartitionMetadataResponse.GetRequestId(), cmd) |
| |
| case pb.BaseCommand_LOOKUP_RESPONSE: |
| lookupResult := cmd.LookupTopicResponse |
| c.checkServerError(lookupResult.Error) |
| c.handleResponse(lookupResult.GetRequestId(), cmd) |
| |
| case pb.BaseCommand_CONSUMER_STATS_RESPONSE: |
| c.handleResponse(cmd.ConsumerStatsResponse.GetRequestId(), cmd) |
| |
| case pb.BaseCommand_GET_LAST_MESSAGE_ID_RESPONSE: |
| c.handleResponse(cmd.GetLastMessageIdResponse.GetRequestId(), cmd) |
| |
| case pb.BaseCommand_GET_TOPICS_OF_NAMESPACE_RESPONSE: |
| c.handleResponse(cmd.GetTopicsOfNamespaceResponse.GetRequestId(), cmd) |
| |
| case pb.BaseCommand_GET_SCHEMA_RESPONSE: |
| c.handleResponse(cmd.GetSchemaResponse.GetRequestId(), cmd) |
| |
| case pb.BaseCommand_GET_OR_CREATE_SCHEMA_RESPONSE: |
| c.handleResponse(cmd.GetOrCreateSchemaResponse.GetRequestId(), cmd) |
| |
| case pb.BaseCommand_ERROR: |
| c.handleResponseError(cmd.GetError()) |
| |
| case pb.BaseCommand_SEND_ERROR: |
| c.handleSendError(cmd.GetSendError()) |
| |
| case pb.BaseCommand_CLOSE_PRODUCER: |
| c.handleCloseProducer(cmd.GetCloseProducer()) |
| |
| case pb.BaseCommand_CLOSE_CONSUMER: |
| c.handleCloseConsumer(cmd.GetCloseConsumer()) |
| |
| case pb.BaseCommand_AUTH_CHALLENGE: |
| c.handleAuthChallenge(cmd.GetAuthChallenge()) |
| |
| case pb.BaseCommand_SEND_RECEIPT: |
| c.handleSendReceipt(cmd.GetSendReceipt()) |
| |
| case pb.BaseCommand_MESSAGE: |
| c.handleMessage(cmd.GetMessage(), headersAndPayload) |
| |
| case pb.BaseCommand_ACK_RESPONSE: |
| c.handleAckResponse(cmd.GetAckResponse()) |
| |
| case pb.BaseCommand_PING: |
| c.handlePing() |
| case pb.BaseCommand_PONG: |
| c.handlePong() |
| case pb.BaseCommand_TC_CLIENT_CONNECT_RESPONSE: |
| c.handleResponse(cmd.TcClientConnectResponse.GetRequestId(), cmd) |
| case pb.BaseCommand_NEW_TXN_RESPONSE: |
| c.handleResponse(cmd.NewTxnResponse.GetRequestId(), cmd) |
| case pb.BaseCommand_ADD_PARTITION_TO_TXN_RESPONSE: |
| c.handleResponse(cmd.AddPartitionToTxnResponse.GetRequestId(), cmd) |
| case pb.BaseCommand_ADD_SUBSCRIPTION_TO_TXN_RESPONSE: |
| c.handleResponse(cmd.AddSubscriptionToTxnResponse.GetRequestId(), cmd) |
| case pb.BaseCommand_END_TXN_RESPONSE: |
| c.handleResponse(cmd.EndTxnResponse.GetRequestId(), cmd) |
| case pb.BaseCommand_ACTIVE_CONSUMER_CHANGE: |
| c.handleActiveConsumerChange(cmd.GetActiveConsumerChange()) |
| |
| default: |
| c.log.Errorf("Received invalid command type: %s", cmd.Type) |
| c.Close() |
| } |
| } |
| |
| func (c *connection) checkServerError(err *pb.ServerError) { |
| if err == nil { |
| return |
| } |
| |
| if *err == pb.ServerError_ServiceNotReady { |
| c.Close() |
| } |
| } |
| |
| func (c *connection) Write(data Buffer) { |
| c.writeRequestsCh <- data |
| } |
| |
| func (c *connection) SendRequest(requestID uint64, req *pb.BaseCommand, |
| callback func(command *pb.BaseCommand, err error)) { |
| c.incomingRequestsWG.Add(1) |
| defer c.incomingRequestsWG.Done() |
| |
| state := c.getState() |
| if state == connectionClosed || state == connectionClosing { |
| callback(req, ErrConnectionClosed) |
| |
| } else { |
| select { |
| case <-c.closeCh: |
| callback(req, ErrConnectionClosed) |
| |
| case c.incomingRequestsCh <- &request{ |
| id: &requestID, |
| cmd: req, |
| callback: callback, |
| }: |
| } |
| } |
| } |
| |
| func (c *connection) SendRequestNoWait(req *pb.BaseCommand) error { |
| c.incomingRequestsWG.Add(1) |
| defer c.incomingRequestsWG.Done() |
| |
| state := c.getState() |
| if state == connectionClosed || state == connectionClosing { |
| return ErrConnectionClosed |
| } |
| |
| select { |
| case <-c.closeCh: |
| return ErrConnectionClosed |
| |
| case c.incomingRequestsCh <- &request{ |
| id: nil, |
| cmd: req, |
| callback: nil, |
| }: |
| return nil |
| } |
| } |
| |
| func (c *connection) internalSendRequest(req *request) { |
| if c.closed() { |
| c.log.Warnf("internalSendRequest failed for connectionClosed") |
| if req.callback != nil { |
| req.callback(req.cmd, ErrConnectionClosed) |
| } |
| } else { |
| c.pendingLock.Lock() |
| if req.id != nil { |
| c.pendingReqs[*req.id] = req |
| } |
| c.pendingLock.Unlock() |
| c.writeCommand(req.cmd) |
| } |
| } |
| |
| func (c *connection) handleResponse(requestID uint64, response *pb.BaseCommand) { |
| request, ok := c.deletePendingRequest(requestID) |
| if !ok { |
| c.log.Warnf("Received unexpected response for request %d of type %s", requestID, response.Type) |
| return |
| } |
| |
| request.callback(response, nil) |
| } |
| |
| func (c *connection) handleResponseError(serverError *pb.CommandError) { |
| requestID := serverError.GetRequestId() |
| |
| request, ok := c.deletePendingRequest(requestID) |
| if !ok { |
| c.log.Warnf("Received unexpected error response for request %d of type %s", |
| requestID, serverError.GetError()) |
| return |
| } |
| |
| errMsg := fmt.Sprintf("server error: %s: %s", serverError.GetError(), serverError.GetMessage()) |
| request.callback(nil, errors.New(errMsg)) |
| } |
| |
| func (c *connection) handleAckResponse(ackResponse *pb.CommandAckResponse) { |
| requestID := ackResponse.GetRequestId() |
| consumerID := ackResponse.GetConsumerId() |
| |
| request, ok := c.deletePendingRequest(requestID) |
| if !ok { |
| c.log.Warnf("AckResponse has complete when receive response! requestId : %d, consumerId : %d", |
| requestID, consumerID) |
| return |
| } |
| |
| if ackResponse.GetMessage() == "" { |
| request.callback(nil, nil) |
| return |
| } |
| |
| errMsg := fmt.Sprintf("ack response error: %s: %s", ackResponse.GetError(), ackResponse.GetMessage()) |
| request.callback(nil, errors.New(errMsg)) |
| } |
| |
| func (c *connection) handleSendReceipt(response *pb.CommandSendReceipt) { |
| producerID := response.GetProducerId() |
| |
| c.listenersLock.RLock() |
| producer, ok := c.listeners[producerID] |
| c.listenersLock.RUnlock() |
| |
| if ok { |
| producer.ReceivedSendReceipt(response) |
| } else { |
| c.log. |
| WithField("producerID", producerID). |
| Warn("Got unexpected send receipt for messageID=%+v", response.MessageId) |
| } |
| } |
| |
| func (c *connection) handleMessage(response *pb.CommandMessage, payload Buffer) { |
| c.log.Debug("Got Message: ", response) |
| consumerID := response.GetConsumerId() |
| if consumer, ok := c.consumerHandler(consumerID); ok { |
| err := consumer.MessageReceived(response, payload) |
| if err != nil { |
| c.log. |
| WithError(err). |
| WithField("consumerID", consumerID). |
| Error("handle message Id: ", response.MessageId) |
| } |
| } else { |
| c.log.WithField("consumerID", consumerID).Warn("Got unexpected message: ", response.MessageId) |
| } |
| } |
| |
| func (c *connection) deletePendingRequest(requestID uint64) (*request, bool) { |
| c.pendingLock.Lock() |
| defer c.pendingLock.Unlock() |
| request, ok := c.pendingReqs[requestID] |
| if ok { |
| delete(c.pendingReqs, requestID) |
| } |
| return request, ok |
| } |
| |
| func (c *connection) findPendingRequest(requestID uint64) (*request, bool) { |
| c.pendingLock.Lock() |
| defer c.pendingLock.Unlock() |
| request, ok := c.pendingReqs[requestID] |
| return request, ok |
| } |
| |
| func (c *connection) failPendingRequests(err error) bool { |
| c.pendingLock.Lock() |
| defer c.pendingLock.Unlock() |
| for id, req := range c.pendingReqs { |
| req.callback(nil, err) |
| delete(c.pendingReqs, id) |
| } |
| return true |
| } |
| |
| func (c *connection) lastDataReceived() time.Time { |
| c.lastDataReceivedLock.Lock() |
| defer c.lastDataReceivedLock.Unlock() |
| t := c.lastDataReceivedTime |
| return t |
| } |
| |
| func (c *connection) setLastDataReceived(t time.Time) { |
| c.lastDataReceivedLock.Lock() |
| defer c.lastDataReceivedLock.Unlock() |
| c.lastDataReceivedTime = t |
| } |
| |
| func (c *connection) sendPing() { |
| c.log.Debug("Sending PING") |
| c.writeCommand(baseCommand(pb.BaseCommand_PING, &pb.CommandPing{})) |
| } |
| |
| func (c *connection) handlePong() { |
| c.log.Debug("Received PONG response") |
| } |
| |
| func (c *connection) handlePing() { |
| c.log.Debug("Responding to PING request") |
| c.writeCommand(baseCommand(pb.BaseCommand_PONG, &pb.CommandPong{})) |
| } |
| |
| func (c *connection) handleAuthChallenge(authChallenge *pb.CommandAuthChallenge) { |
| c.log.Debugf("Received auth challenge from broker: %s", authChallenge.GetChallenge().GetAuthMethodName()) |
| |
| // Get new credentials from the provider |
| authData, err := c.auth.GetData() |
| if err != nil { |
| c.log.WithError(err).Warn("Failed to load auth credentials") |
| c.Close() |
| return |
| } |
| |
| // Brokers expect authData to be not nil |
| if authData == nil { |
| authData = []byte{} |
| } |
| |
| cmdAuthResponse := &pb.CommandAuthResponse{ |
| ProtocolVersion: proto.Int32(PulsarProtocolVersion), |
| ClientVersion: proto.String(ClientVersionString), |
| Response: &pb.AuthData{ |
| AuthMethodName: proto.String(c.auth.Name()), |
| AuthData: authData, |
| }, |
| } |
| |
| c.writeCommand(baseCommand(pb.BaseCommand_AUTH_RESPONSE, cmdAuthResponse)) |
| } |
| |
| func (c *connection) handleSendError(sendError *pb.CommandSendError) { |
| c.log.Warnf("Received send error from server: [%v] : [%s]", sendError.GetError(), sendError.GetMessage()) |
| |
| producerID := sendError.GetProducerId() |
| |
| switch sendError.GetError() { |
| case pb.ServerError_NotAllowedError: |
| _, ok := c.deletePendingProducers(producerID) |
| if !ok { |
| c.log.Warnf("Received unexpected error response for request %d of type %s", |
| producerID, sendError.GetError()) |
| return |
| } |
| |
| c.log.Warnf("server error: %s: %s", sendError.GetError(), sendError.GetMessage()) |
| case pb.ServerError_TopicTerminatedError: |
| _, ok := c.deletePendingProducers(producerID) |
| if !ok { |
| c.log.Warnf("Received unexpected error response for producer %d of type %s", |
| producerID, sendError.GetError()) |
| return |
| } |
| c.log.Warnf("server error: %s: %s", sendError.GetError(), sendError.GetMessage()) |
| default: |
| // By default, for transient error, let the reconnection logic |
| // to take place and re-establish the produce again |
| c.Close() |
| } |
| } |
| |
| func (c *connection) deletePendingProducers(producerID uint64) (ConnectionListener, bool) { |
| c.listenersLock.Lock() |
| producer, ok := c.listeners[producerID] |
| if ok { |
| delete(c.listeners, producerID) |
| } |
| c.listenersLock.Unlock() |
| |
| return producer, ok |
| } |
| |
| func (c *connection) handleCloseConsumer(closeConsumer *pb.CommandCloseConsumer) { |
| consumerID := closeConsumer.GetConsumerId() |
| c.log.Infof("Broker notification of Closed consumer: %d", consumerID) |
| |
| if consumer, ok := c.consumerHandler(consumerID); ok { |
| consumer.ConnectionClosed(closeConsumer) |
| c.DeleteConsumeHandler(consumerID) |
| } else { |
| c.log.WithField("consumerID", consumerID).Warnf("Consumer with ID not found while closing consumer") |
| } |
| } |
| |
| func (c *connection) handleActiveConsumerChange(consumerChange *pb.CommandActiveConsumerChange) { |
| consumerID := consumerChange.GetConsumerId() |
| isActive := consumerChange.GetIsActive() |
| if consumer, ok := c.consumerHandler(consumerID); ok { |
| consumer.ActiveConsumerChanged(isActive) |
| } else { |
| c.log.WithField("consumerID", consumerID).Warnf("Consumer not found while active consumer change") |
| } |
| } |
| |
| func (c *connection) handleCloseProducer(closeProducer *pb.CommandCloseProducer) { |
| c.log.Infof("Broker notification of Closed producer: %d", closeProducer.GetProducerId()) |
| producerID := closeProducer.GetProducerId() |
| |
| producer, ok := c.deletePendingProducers(producerID) |
| // did we find a producer? |
| if ok { |
| producer.ConnectionClosed(closeProducer) |
| } else { |
| c.log.WithField("producerID", producerID).Warn("Producer with ID not found while closing producer") |
| } |
| } |
| |
| func (c *connection) RegisterListener(id uint64, listener ConnectionListener) error { |
| // do not add if connection is closed |
| if c.closed() { |
| c.log.Warnf("Connection closed unable register listener id=%+v", id) |
| return errUnableRegisterListener |
| } |
| |
| c.listenersLock.Lock() |
| defer c.listenersLock.Unlock() |
| |
| c.listeners[id] = listener |
| return nil |
| } |
| |
| func (c *connection) UnregisterListener(id uint64) { |
| c.listenersLock.Lock() |
| defer c.listenersLock.Unlock() |
| |
| delete(c.listeners, id) |
| } |
| |
| func (c *connection) ResetLastActive() { |
| c.Lock() |
| defer c.Unlock() |
| c.lastActive = time.Now() |
| } |
| |
| func (c *connection) isIdle() bool { |
| { |
| c.pendingLock.Lock() |
| defer c.pendingLock.Unlock() |
| if len(c.pendingReqs) != 0 { |
| return false |
| } |
| } |
| |
| { |
| c.listenersLock.RLock() |
| defer c.listenersLock.RUnlock() |
| if len(c.listeners) != 0 { |
| return false |
| } |
| } |
| |
| { |
| c.consumerHandlersLock.Lock() |
| defer c.consumerHandlersLock.Unlock() |
| if len(c.consumerHandlers) != 0 { |
| return false |
| } |
| } |
| |
| if len(c.incomingRequestsCh) != 0 || len(c.writeRequestsCh) != 0 { |
| return false |
| } |
| return true |
| } |
| |
| func (c *connection) CheckIdle(maxIdleTime time.Duration) bool { |
| // We don't need to lock here because this method should only be |
| // called in a single goroutine of the connectionPool |
| if !c.isIdle() { |
| c.lastActive = time.Now() |
| } |
| return time.Since(c.lastActive) > maxIdleTime |
| } |
| |
| // Close closes the connection by |
| // closing underlying socket connection and closeCh. |
| // This also triggers callbacks to the ConnectionClosed listeners. |
| func (c *connection) Close() { |
| c.closeOnce.Do(func() { |
| c.Lock() |
| cnx := c.cnx |
| c.Unlock() |
| c.changeState(connectionClosed) |
| |
| if cnx != nil { |
| _ = cnx.Close() |
| } |
| |
| close(c.closeCh) |
| |
| listeners := make(map[uint64]ConnectionListener) |
| c.listenersLock.Lock() |
| for id, listener := range c.listeners { |
| listeners[id] = listener |
| delete(c.listeners, id) |
| } |
| c.listenersLock.Unlock() |
| |
| consumerHandlers := make(map[uint64]ConsumerHandler) |
| c.consumerHandlersLock.Lock() |
| for id, handler := range c.consumerHandlers { |
| consumerHandlers[id] = handler |
| delete(c.consumerHandlers, id) |
| } |
| c.consumerHandlersLock.Unlock() |
| |
| // notify producers connection closed |
| for _, listener := range listeners { |
| listener.ConnectionClosed(nil) |
| } |
| |
| // notify consumers connection closed |
| for _, handler := range consumerHandlers { |
| handler.ConnectionClosed(nil) |
| } |
| |
| c.metrics.ConnectionsClosed.Inc() |
| }) |
| } |
| |
| func (c *connection) changeState(state connectionState) { |
| // The lock is held here because we need setState() and cond.Broadcast() to be |
| // an atomic operation from the point of view of waitUntilReady(). |
| c.Lock() |
| defer c.Unlock() |
| |
| c.setState(state) |
| c.cond.Broadcast() |
| } |
| |
| func (c *connection) getState() connectionState { |
| return connectionState(c.state.Load()) |
| } |
| |
| func (c *connection) setState(state connectionState) { |
| c.state.Store(int32(state)) |
| } |
| |
| func (c *connection) closed() bool { |
| return connectionClosed == c.getState() |
| } |
| |
| func (c *connection) getTLSConfig() (*tls.Config, error) { |
| tlsConfig := &tls.Config{ |
| InsecureSkipVerify: c.tlsOptions.AllowInsecureConnection, |
| CipherSuites: c.tlsOptions.CipherSuites, |
| MinVersion: c.tlsOptions.MinVersion, |
| MaxVersion: c.tlsOptions.MaxVersion, |
| } |
| |
| if c.tlsOptions.TrustCertsFilePath != "" { |
| caCerts, err := os.ReadFile(c.tlsOptions.TrustCertsFilePath) |
| if err != nil { |
| return nil, err |
| } |
| |
| tlsConfig.RootCAs = x509.NewCertPool() |
| ok := tlsConfig.RootCAs.AppendCertsFromPEM(caCerts) |
| if !ok { |
| return nil, errors.New("failed to parse root CAs certificates") |
| } |
| } |
| |
| if c.tlsOptions.ValidateHostname { |
| if c.tlsOptions.ServerName != "" { |
| tlsConfig.ServerName = c.tlsOptions.ServerName |
| } else { |
| tlsConfig.ServerName = c.physicalAddr.Hostname() |
| } |
| c.log.Debugf("getTLSConfig(): setting tlsConfig.ServerName = %+v", tlsConfig.ServerName) |
| } |
| |
| if c.tlsOptions.CertFile != "" && c.tlsOptions.KeyFile != "" { |
| cert, err := tls.LoadX509KeyPair(c.tlsOptions.CertFile, c.tlsOptions.KeyFile) |
| if err != nil { |
| return nil, errors.New(err.Error()) |
| } |
| tlsConfig.Certificates = []tls.Certificate{cert} |
| } |
| |
| cert, err := c.auth.GetTLSCertificate() |
| if err != nil { |
| return nil, err |
| } |
| |
| if cert != nil { |
| tlsConfig.Certificates = []tls.Certificate{*cert} |
| } |
| |
| return tlsConfig, nil |
| } |
| |
| func (c *connection) AddConsumeHandler(id uint64, handler ConsumerHandler) error { |
| // do not add if connection is closed |
| if c.closed() { |
| c.log.Warnf("Closed connection unable add consumer with id=%+v", id) |
| return errUnableAddConsumeHandler |
| } |
| |
| c.consumerHandlersLock.Lock() |
| defer c.consumerHandlersLock.Unlock() |
| c.consumerHandlers[id] = handler |
| return nil |
| } |
| |
| func (c *connection) DeleteConsumeHandler(id uint64) { |
| c.consumerHandlersLock.Lock() |
| defer c.consumerHandlersLock.Unlock() |
| delete(c.consumerHandlers, id) |
| } |
| |
| func (c *connection) consumerHandler(id uint64) (ConsumerHandler, bool) { |
| c.consumerHandlersLock.RLock() |
| defer c.consumerHandlersLock.RUnlock() |
| h, ok := c.consumerHandlers[id] |
| return h, ok |
| } |
| |
| func (c *connection) ID() string { |
| return fmt.Sprintf("%s -> %s", c.cnx.LocalAddr(), c.cnx.RemoteAddr()) |
| } |
| |
| func (c *connection) GetMaxMessageSize() int32 { |
| return c.maxMessageSize |
| } |
