Interface function to send and receive decoded bytes
        Retries the connect [self.retries] times
553550
def get(self, command=None):
        

        try:
            assert self.telnet
            with self.lock:
                if command:
                    if not command.endswith('\r\n'):
                        command += '\r\n'
                    LOGGER.debug('%s: sending %r', self.host, command)
                    self.telnet.write(command.encode())

                res = self.telnet.read_until('\r\n'.encode()).decode()
                LOGGER.debug('%s: received %r', self.host, res)
                if res.split()[0] not in ('100', '250'):
                    LOGGER.warn('command error: %r', res)
                    return None
                return res.split()[1]

        except (AssertionError):
            self.connect()

        except (EOFError, socket.gaierror, BrokenPipeError):
            LOGGER.error("Cannot get answer from %s (%d)",
                         self.host, self.retries)
            if self.retries > 0:
                self.retries -= 1
                self.connect()
                return self.get(command)
            else:
                self.retries = Netio.MAX_RETRIES
                return None
775746
function() {
        // Send request
        var sendRequest = function() {
          debug('Calling: %s, retry: %s', entry.name, attempts - 1);
          // Make request and handle response or error
          return makeRequest(
            that,
            entry.method,
            url,
            payload,
            query
          ).then(function(res) {
            // If request was successful, accept the result
            debug('Success calling: %s, (%s retries)',
              entry.name, attempts - 1);
            if (monitor) {
              var d = process.hrtime(start);
              monitor.measure([entry.name, 'success'], d[0] * 1000 + d[1] / 1000000);
              monitor.count([entry.name, 'success']);
            }
            if (!_.includes(res.headers['content-type'], 'application/json') || !res.body) {
              debug('Empty response from server: call: %s, method: %s', entry.name, entry.method);
              return undefined;
            }
            return res.body;
          }, function(err) {
            // If we got a response we read the error code from the response
            var res = err.response;
            if (res) {
              // Decide if we should retry
              if (attempts <= that._options.retries &&
                  res.status >= 500 &&  // Check if it's a 5xx error
                  res.status < 600) {
                debug('Error calling: %s now retrying, info: %j',
                  entry.name, res.body);
                return retryRequest();
              }
              // If not retrying, construct error object and reject
              debug('Error calling: %s NOT retrying!, info: %j',
                entry.name, res.body);
              var message = 'Unknown Server Error';
              if (res.status === 401) {
                message = 'Authentication Error';
              }
              if (res.status === 500) {
                message = 'Internal Server Error';
              }
              err = new Error(res.body.message || message);
              err.body = res.body;
              err.code = res.body.code || 'UnknownError';
              err.statusCode = res.status;
              if (monitor) {
                var d = process.hrtime(start);

                var state = 'client-error';
                if (res.statusCode >= 500) {
                  state = 'server-error';
                }
                monitor.measure([entry.name, state], d[0] * 1000 + d[1] / 1000000);
                monitor.count([entry.name, state]);
              }
              throw err;
            }

            // Decide if we should retry
            if (attempts <= that._options.retries) {
              debug('Request error calling %s (retrying), err: %s, JSON: %s',
                entry.name, err, err);
              return retryRequest();
            }
            debug('Request error calling %s NOT retrying!, err: %s, JSON: %s',
              entry.name, err, err);
            if (monitor) {
              var d = process.hrtime(start);
              monitor.measure([entry.name, 'connection-error'], d[0] * 1000 + d[1] / 1000000);
              monitor.count([entry.name, 'connection-error']);
            }
            throw err;
          });
        };

        // Increment attempt count, but track how many we had before.
        attempts += 1;

        // If this is the first retry, ie we haven't retried yet, we make the
        // request immediately
        if (attempts === 1) {
          return sendRequest();
        } else {
          var delay;
          // First request is attempt = 1, so attempt = 2 is the first retry
          // we subtract one to get exponents: 1, 2, 3, 4, 5, ...
          delay = Math.pow(2, attempts - 1) * that._options.delayFactor;
          // Apply randomization factor
          var rf = that._options.randomizationFactor;
          delay *= Math.random() * 2 * rf + 1 - rf;
          // Always limit with a maximum delay
          delay = Math.min(delay, that._options.maxDelay);
          // Sleep then send the request
          return new Promise(function(accept) {
            setTimeout(accept, delay);
          }).then(function() {
            return sendRequest();
          });
        }
      }
183870
function(func, opts) {
	opts = Object.assign({
		chunk_size: 25,
		retry: 2,
		retryDelay: 100,
		concurrency: 2,
		concurrency_delay: 100,
		combine: false,
		data_size: null
	}, opts || {});
	// logger.log("opts are ", opts);
	var records = [];
	var calls = 0;
	var completedCalls = 0;
	var requestEnd = false;

	var retries = 0;
	var errors = 0;
	var hadErrors = false;
	var batches = 0;
	var delaying = false;

	function sendAvailable() {
		var sendSize;
		if (records.length > 0 && retries <= opts.retry && completedCalls == calls && !delaying) {
			if (errors == 0) { //let's reset because last round completed successfully
				batches++;
				if (!hadErrors) {
					retries = 0;
				}
				hadErrors = false;
				logger.log(`-------------------New Batch #${batches}----------------`);
				if (opts.chunk_size < 10 || opts.concurrency > 25) {
					logger.log(`chunking ${opts.chunk_size} - ${opts.concurrency} times`);

				}
			} else {
				logger.log(`-------------------Retrying: ${errors} records failed, retrying in ${opts.retryDelay * retries}ms, retry #${opts.retry - (opts.retry - retries) + 1}----------------`);
				retries++;
				errors = 0;
				hadErrors = true;
				delaying = true;
				setTimeout(function() {
					delaying = false;
					sendAvailable();
				}, opts.retryDelay * retries);
				return;
			}
			if (retries > opts.retry) {
				checkDone();
				return;
			}
			while (records.length > 0 && completedCalls > calls - opts.concurrency) {
				calls++;
				var dataSizeBased = false;
				if (opts.data_size) {
					sendSize = 0;
					var runningSize = 0;
					for (var i = 0; i < opts.chunk_size && i < records.length; i++) {
						var r = records[i];
						runningSize += r.size;
						if (runningSize > opts.data_size) {
							dataSizeBased = true;
							break;
						}
						sendSize++;
					}
				} else {
					sendSize = opts.chunk_size;
				}
				if (opts.combine) {
					var items = records.splice(0, sendSize);
					var toProcess = [];
					var size = 0;
					var groupStart = 0;

					for (var i = 0; i < items.length; i++) {
						var item = items[i];
						if (item.size + size >= opts.record_size) {
							logger.log(`grouping items from ${groupStart+1} to ${i} of ${items.length} of size: ${size}`);
							toProcess.push(items.slice(groupStart, i).map((e) => {
								return e.record;
							}).join(''));
							groupStart = i;
							size = item.size;
						} else {
							size += item.size;
						}
					}
					if (groupStart != items.length) {
						logger.log(`grouping items from ${groupStart+1} to ${items.length} of ${items.length} of size: ${size}`);
						toProcess.push(items.slice(groupStart, items.length).map((e) => {
							return e.record;
						}).join(''));
					}

				} else {
					var toProcess = records.splice(0, sendSize).map(function(e) {
						return e.record;
					});
				}

				if (toProcess.length > 0) {
					if (opts.chunk_size >= 10 && opts.concurrency <= 25) {
						logger.log(`chunking ${toProcess.length} records (${dataSizeBased?'Data Size':'Count Size'})`);
					}
					func(toProcess, function(err, unprocessedItems) {
						if (err) {
							logger.log(`Records not processed, ${unprocessedItems.length}`);

							process.nextTick(function() {
								//Don't want to add the records or change completed calls until after the current While loop is done...otherwise a nasty infinite loop could happen
								completedCalls++;
								records = unprocessedItems.map(function(e) {
									var size;
									if (!size) {
										if (typeof e === "string") {
											size = Buffer.byteLength(e);
										} else {
											size = Buffer.byteLength(JSON.stringify(e));
										}
									}
									return {
										size: size,
										record: e
									};
								}).concat(records);
								errors += unprocessedItems.length;
								setTimeout(sendAvailable, opts.concurrency_delay);
							});
						} else if (records.length) {
							completedCalls++;
							setTimeout(sendAvailable, opts.concurrency_delay);
						} else {
							completedCalls++;
							sendAvailable();
						}
					});
				} else {
					completedCalls++;
				}
			}
		} else {
			checkDone();
		}
	}

	function checkDone() {
		if (requestEnd !== false && completedCalls == calls && (records.length == 0 || retries >= opts.retry)) {
			if (records.length > 0) {
				requestEnd("Cannot process all the entries", records.length);
				requestEnd = false;
			} else {
				requestEnd(null, []);
				requestEnd = false;
			}
		}
	}

	return {
		add: function(item) {
			requestEnd = false;
			var size;
			if (!size) {
				if (typeof item === "string") {
					size = Buffer.byteLength(item);
				} else {
					size = Buffer.byteLength(JSON.stringify(item));
				}
			}

			if (opts.record_size && size > opts.record_size) {
				logger.error("record size is too large", size, opts.record_size);
			} else if (opts.data_size && size > opts.data_size) {
				logger.error("data size is too large");
			} else {
				records.push({
					size: size,
					record: item
				});
				if (records.length >= opts.chunk_size * opts.concurrency) {
					sendAvailable();
				}
			}
		},
		end: function(callback) {
			requestEnd = callback;
			sendAvailable();
		}
	};
}
994315
def send_data(str)
      if !connected?
        if !@socket.nil? && @reconnect == false
          raise Munin::ConnectionError, "Not connected."
        else
          open
        end
      end

      begin
        with_timeout { @socket.puts("#{str.strip}\n") }
      rescue Timeout::Error
        raise Munin::ConnectionError, "Timed out on #{@host} trying to send."
      end
    end
314637
function retryOrError(){ 
					if (bRetrying)
						throw Error("unit hash verification failed, unit: "+unit+", objUnit: "+JSON.stringify(objUnit));
					console.log("unit hash verification failed, will retry");
					return setTimeout(function(){
						readJointDirectly(conn, unit, callbacks, true);
					}, 60*1000);
				}
328872
def retryable(fn, protocol=None):
    
    def wrapper(self, *args, **kwargs):
        pool = self._choose_pool(protocol)

        def thunk(transport):
            return fn(self, transport, *args, **kwargs)

        return self._with_retries(pool, thunk)

    wrapper.__doc__ = fn.__doc__
    wrapper.__repr__ = fn.__repr__

    return wrapper
793492
async def on_connect(self, connection):
        ""
        # NOTE: for python3, we can't pass bytestrings as keyword arguments
        # so we need to decode channel/pattern names back to str strings
        # before passing them to [p]subscribe.
        if self.channels:
            channels = {}
            for k, v in iteritems(self.channels):
                if not self.decode_responses:
                    k = k.decode(self.encoding)
                channels[k] = v
            await self.subscribe(**channels)
        if self.patterns:
            patterns = {}
            for k, v in iteritems(self.patterns):
                if not self.decode_responses:
                    k = k.decode(self.encoding)
                patterns[k] = v
            await self.psubscribe(**patterns)
863190
def send(self, data):
        

        if self.sock is None:
            if self.auto_open:
                self.connect()
            else:
                raise NotConnected()

        if self.debuglevel > 0:
            print("send:", repr(data))
        blocksize = 8192
        # Python 2.7 array objects have a read method which is incompatible
        # with the 2-arg calling syntax below.
        if hasattr(data, "read") and not isinstance(data, array):
            if self.debuglevel > 0:
                print("sendIng a read()able")
            encode = False
            try:
                mode = data.mode
            except AttributeError:
                # io.BytesIO and other file-like objects don't have a `mode`
                # attribute.
                pass
            else:
                if "b" not in mode:
                    encode = True
                    if self.debuglevel > 0:
                        print("encoding file using iso-8859-1")
            while 1:
                datablock = data.read(blocksize)
                if not datablock:
                    break
                if encode:
                    datablock = datablock.encode("iso-8859-1")
                self.sock.sendall(datablock)
            return
        try:
            self.sock.sendall(data)
        except TypeError:
            if isinstance(data, collections.Iterable):
                for d in data:
                    self.sock.sendall(d)
            else:
                raise TypeError("data should be a bytes-like object "
                                "or an iterable, got %r" % type(data))
231742
def pre_send(self, request_params):
        """"""
        for adapter in itervalues(self.adapters):
            adapter.max_retries = request_params.get('max_retries', 0)

        return request_params
676331
def retrbinary(cmd, blocksize, rest_offset = nil) # :yield: data
      synchronize do
        with_binary(true) do
          begin
            conn = transfercmd(cmd, rest_offset)
            loop do
              data = conn.read(blocksize)
              break if data == nil
              yield(data)
            end
            conn.shutdown(Socket::SHUT_WR)
            conn.read_timeout = 1
            conn.read
          ensure
            conn.close if conn
          end
          voidresp
        end
      end
    end
484210
def recv(self, bufsize, **kws):
        
        return Recv(self, bufsize, timeout=self._timeout, **kws)
1314680
def _retryable_read(self, func, read_pref, session, address=None,
                        retryable=True, exhaust=False):
        
        retryable = (retryable and
                     self.retry_reads
                     and not (session and session._in_transaction))
        last_error = None
        retrying = False

        while True:
            try:
                server = self._select_server(
                    read_pref, session, address=address)
                if not server.description.retryable_reads_supported:
                    retryable = False
                with self._slaveok_for_server(read_pref, server, session,
                                              exhaust=exhaust) as (sock_info,
                                                                   slave_ok):
                    if retrying and not retryable:
                        # A retry is not possible because this server does
                        # not support retryable reads, raise the last error.
                        raise last_error
                    return func(session, server, sock_info, slave_ok)
            except ServerSelectionTimeoutError:
                if retrying:
                    # The application may think the write was never attempted
                    # if we raise ServerSelectionTimeoutError on the retry
                    # attempt. Raise the original exception instead.
                    raise last_error
                # A ServerSelectionTimeoutError error indicates that there may
                # be a persistent outage. Attempting to retry in this case will
                # most likely be a waste of time.
                raise
            except ConnectionFailure as exc:
                if not retryable or retrying:
                    raise
                retrying = True
                last_error = exc
            except OperationFailure as exc:
                if not retryable or retrying:
                    raise
                if exc.code not in helpers._RETRYABLE_ERROR_CODES:
                    raise
                retrying = True
                last_error = exc
912503
def retry(func):
    
    def retried_func(*args, **kwargs):
        max_tries = 3
        tries = 0
        while True:
            try:
                resp = func(*args, **kwargs)

            except requests.exceptions.ConnectionError as exc:
                exc.msg = "Connection error for session; exiting"
                raise exc

            except requests.exceptions.HTTPError as exc:
                exc.msg = "HTTP error for session; exiting"
                raise exc

            if resp.status_code != 200 and tries < max_tries:
                logger.warning("retrying request; current status code: {}"
                               .format(resp.status_code))
                tries += 1
                # mini exponential backoff here.
                time.sleep(tries ** 2)
                continue

            break

        if resp.status_code != 200:
            error_message = resp.json()["error"]["message"]
            logger.error("HTTP Error code: {}: {}".format(resp.status_code, error_message))
            logger.error("Rule payload: {}".format(kwargs["rule_payload"]))
            raise requests.exceptions.HTTPError

        return resp

    return retried_func
168740
public boolean sendTCPString(String message, int retries)
    {
        Log(Level.FINE, "About to send: " + message);
        byte[] bytes = message.getBytes();
        return sendTCPBytes(bytes, retries);
    }
305061
def write(self, data):
        ''
        return self._with_retry(
                functools.partial(self._sslobj.write, data),
                self.gettimeout())
945630
def receive(self, decode=True):
        
        payload = self.socket.recv()
        payload = self.verify(payload)
        if decode:
            payload = self.decode(payload)
        return payload
696319
def retry(self, f, *args, **kwargs):
        
        backoff = random.random() / 100  # 5ms on average
        for _ in range(self.tries - 1):
            try:
                return f(*args, **kwargs)
            except NetworkError:
                time.sleep(backoff)
                backoff *= 2
        return f(*args, **kwargs)