{* ../../docs_src/stream_data/tutorial001_py310.py ln[32:35] hl[33] *}

這也意味著使用 `StreamingResponse` 時，你擁有自由與責任，需依需求自行產生並編碼要傳送的位元組資料，與型別註解無關。 🤓

### 串流位元組 { #stream-bytes }

一個主要用例是串流 `bytes` 而非字串，當然可以這麼做。

{* ../../docs_src/stream_data/tutorial001_py310.py ln[44:47] hl[47] *}

## 自訂 `PNGStreamingResponse` { #a-custom-pngstreamingresponse }

package bytes_test

import (
	. "bytes"
	"fmt"
	"testing"
)

var compareTests = []struct {
	a, b []byte
	i    int
}{
	{[]byte(""), []byte(""), 0},
	{[]byte("a"), []byte(""), 1},
	{[]byte(""), []byte("a"), -1},
	{[]byte("abc"), []byte("abc"), 0},
	{[]byte("abd"), []byte("abc"), 1},
	{[]byte("abc"), []byte("abd"), -1},
	{[]byte("ab"), []byte("abc"), -1},
	{[]byte("abc"), []byte("ab"), 1},

SuccessExitStatus=143

# Specifies the maximum file descriptor number that can be opened by this process
LimitNOFILE=${packaging.os.max.open.files}

# Specifies the maximum number of bytes of memory that may be locked into RAM
# Set to "infinity" if you use the 'bootstrap.mlockall: true' option
# in fess.yml and 'MAX_LOCKED_MEMORY=unlimited' in ${packaging.env.file}
#LimitMEMLOCK=infinity

class ContentSizeLimitMiddleware:
    """Content size limiting middleware for ASGI applications
    Args:
      app (ASGI application): ASGI application
      max_content_size (optional): the maximum content size allowed in bytes, None for no limit
    """

    def __init__(self, app: APIRouter, max_content_size: int | None = None):
        self.app = app
        self.max_content_size = max_content_size

    def receive_wrapper(self, receive):

			expectedCode:    304,
		},
	}
	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			recorder := httptest.NewRecorder()
			request := httptest.NewRequest(http.MethodHead, "/bucket/a", bytes.NewReader([]byte{}))
			request.Header.Set(xhttp.IfNoneMatch, tc.ifNoneMatch)
			request.Header.Set(xhttp.IfModifiedSince, tc.ifModifiedSince)
			request.Header.Set(xhttp.IfMatch, tc.ifMatch)

//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package main

import (
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"strings"
)

var (
	keyFile  string
	certFile string
)

func init() {

   */
  var peerSettings = DEFAULT_SETTINGS

  /** The bytes consumed and acknowledged by the application. */
  val readBytes: WindowCounter = WindowCounter(streamId = 0)

  /** The total number of bytes produced by the application. */
  var writeBytesTotal = 0L
    private set

  /** The total number of bytes permitted to be produced according to `WINDOW_UPDATE` frames. */

            final byte[] mechanismListMIC) {
        setMechanisms(mechanisms);
        setContextFlags(contextFlags);
        setMechanismToken(mechanismToken);
        setMechanismListMIC(mechanismListMIC);
    }

    /**
     * Constructs a NegTokenInit by parsing the provided token bytes
     * @param token the SPNEGO token bytes to parse

以下是一個更完整的範例。

使用一個依賴來檢查使用者名稱與密碼是否正確。

為此，使用 Python 標準模組 [`secrets`](https://docs.python.org/3/library/secrets.html) 來比對使用者名稱與密碼。

`secrets.compare_digest()` 需要接收 `bytes`，或是只包含 ASCII 字元（英文字符）的 `str`。這表示它無法處理像 `á` 這樣的字元，例如 `Sebastián`。

為了處理這點，我們會先將 `username` 與 `password` 以 UTF-8 編碼成 `bytes`。

接著我們可以使用 `secrets.compare_digest()` 來確認 `credentials.username` 等於 `"stanleyjobson"`，而 `credentials.password` 等於 `"swordfish"`。

		NodeCount:             numNodes,
	}
	go m.trackEWMA()
	return m
}

func (m *Monitor) updateMeasurement(opts BucketOptions, bytes uint64) {
	m.mlock.Lock()
	defer m.mlock.Unlock()

	tm, ok := m.bucketsMeasurement[opts]
	if !ok {
		tm = &bucketMeasurement{}
	}
	tm.incrementBytes(bytes)
	m.bucketsMeasurement[opts] = tm
}

// SelectionFunction for buckets
type SelectionFunction func(bucket string) bool