wandenberg / nginx-push-stream-module

A pure stream http push technology for your Nginx setup. Comet made easy and really scalable.
Other
2.22k stars 295 forks source link

the response format confused.. #45

Closed Terry-Mao closed 11 years ago

Terry-Mao commented 12 years ago

the http response get 200 status, everythng gonaa all right , but somethimes the response like the format:

chunk-size \r\n chunk-data\r\n .....

0\r\n

i know this is the http 1.1 transfer-encoding : chunked

but the issue is that i got the reponse without chunk-size sometimes

like :

chunk-data \r\n chunk-data \r\n.

it's hard to parse the response for client.

my nginx config file:

push_stream_shared_memory_size 300M; push_stream_shared_memory_cleanup_objects_ttl 30s; push_stream_message_ttl 72h; push_stream_max_messages_stored_per_channel 20; chunked_transfer_encoding on; chunkin on;

server { listen 61.147.75.31:80; listen 112.84.189.31:80; charset utf-8; access_log /data/logs/nginx/notify.log; keepalive_timeout 300;

location ~ /sub/(.*) {
    set $push_stream_channels_path  $1;
    push_stream_subscriber  long-polling;
    push_stream_authorized_channels_only    off;
    push_stream_content_type    text/plain;
    push_stream_subscriber_connection_ttl   300s;
    push_stream_longpolling_connection_ttl  300s;
}

error_page 411 = @my_411_error;

location @my_411_error {
    chunkin_resume;
}

}

server { listen 192.168.1.15:80; charset utf-8; access_log /data/logs/nginx/notify.log; keepalive_timeout 300;

    location /pub {
    push_stream_publisher admin;
    set $push_stream_channel_id $arg_channel;
    push_stream_store_messages  on;
    push_stream_keepalive   on;
    }

error_page 411 = @my_411_error;
location @my_411_error {
    chunkin_resume;
}

}

wandenberg commented 12 years ago

Hi,

which version of nginx and push-stream module are you using? Can you send me a way how to reproduce the problem?

Which client are you using? The responsibility of parse the response must be of the client, not yours :)

Regards, Wandenberg

On Mon, Aug 20, 2012 at 2:45 AM, Terry.Mao notifications@github.com wrote:

the http response get 200 status, everythng gonaa all right , but somethimes the response like the format:

chunk-size \r\n chunk-data\r\n .....

0\r\n

i know this is the http 1.1 transfer-encoding : chunked

but the issue is that i got the reponse without chunk-size sometimes

like :

chunk-data \r\n chunk-data \r\n.

it's hard to parse the response for client.

my nginx config file:

push_stream_shared_memory_size 300M; push_stream_shared_memory_cleanup_objects_ttl 30s; push_stream_message_ttl 72h; push_stream_max_messages_stored_per_channel 20; chunked_transfer_encoding on; chunkin on;

server { listen 61.147.75.31:80; listen 112.84.189.31:80; charset utf-8; access_log /data/logs/nginx/notify.log; keepalive_timeout 300;

location ~ /sub/(.*) { set $push_stream_channels_path $1; push_stream_subscriber long-polling; push_stream_authorized_channels_only off; push_stream_content_type text/plain; push_stream_subscriber_connection_ttl 300s; push_stream_longpolling_connection_ttl 300s; }

error_page 411 = @my_411_error;

location @my_411_error { chunkin_resume; }

}

server { listen 192.168.1.15:80; charset utf-8; access_log /data/logs/nginx/notify.log; keepalive_timeout 300;

location /pub {
push_stream_publisher admin;
set $push_stream_channel_id $arg_channel;
push_stream_store_messages  on;
push_stream_keepalive   on;
}

error_page 411 = @my_411_error; location @my_411_error { chunkin_resume; }

}

— Reply to this email directly or view it on GitHubhttps://github.com/wandenberg/nginx-push-stream-module/issues/45.

MysqlProxy commented 12 years ago

nginx : nginx-1.2.3 push-stream : nginx-push-stream-module-28d9df7

i post data to nginx by java (the code is not on my computer now, T.T, tomorrow i'll paste it) i get data from nginx by android (some java code ,as above)

MysqlProxy commented 12 years ago

it's Random, i also use browser like Chorme or FF.

i visit notify.weikan.cn/sub/1 then suspend till i post data curl -v -X POST notify.weikan.cn/pub?channel=1 -d "test'

sometimes i got :

test

sometimes i got:

4 test

0

Terry-Mao commented 12 years ago

i post msg use java code :

public static boolean post(String url, String msg) {
    HttpPost post = null;
    InputStreamEntity reqEntity = null;
    HttpEntity resEntity = null;
    BufferedReader http_reader = null;
    StringBuffer sResult = null;

    try {
        post = new HttpPost(url);
        /*
         * reqEntity = new StringEntity(msg, "application/json", "UTF-8");
         * reqEntity.setContentEncoding(new BasicHeader(HTTP.CONTENT_TYPE,
         * "application/json")); reqEntity.setChunked(false);
         */

        byte[] reqData = msg.getBytes("UTF-8");
        reqEntity = new InputStreamEntity(
                new ByteArrayInputStream(reqData), reqData.length);
        reqEntity.setChunked(true);
        post.setEntity((HttpEntity) reqEntity);
        PushFeeder.logger.info("Content-Length : "
                + reqEntity.getContentLength());
        HttpResponse res = client.execute(post);

        if (HttpStatus.SC_OK == res.getStatusLine().getStatusCode()) {
            resEntity = res.getEntity();
            if (resEntity != null) {
                http_reader = new BufferedReader(new InputStreamReader(
                        resEntity.getContent()));

                String line = null;
                sResult = new StringBuffer();
                while ((line = http_reader.readLine()) != null) {
                    if (sResult.length() > 0) {
                        sResult.append("\n");
                    }
                    sResult.append(line);
                }

                PushFeeder.logger.info(sResult.toString());
            }

            return true;
        } else {
            PushFeeder.logger.info("url : " + url + ", ret : "
                    + res.getStatusLine().getStatusCode());
        }

    } catch (Exception e) {
        PushFeeder.logger.error("", e);
    } finally {
        if (http_reader != null) {
            try {
                http_reader.close();
            } catch (IOException e) {
                PushFeeder.logger.error("", e);
            }
        }
    }

    return false;
}

my android client use java code to get response:

public class TestSubcribe { private String mLastModify; private String mETag;

public void run() {
    String url = "http://****/PUBLIC";
    subcribe(url);
}

private HttpURLConnection getConnection(String url) throws MalformedURLException, IOException {
    HttpURLConnection connection = (HttpURLConnection)(new URL(url.toString()).openConnection());
    connection.setConnectTimeout(30 * 60 * 1000);
    connection.setReadTimeout(30 * 60 * 1000);
    connection.addRequestProperty("If-Modified-Since", mLastModify);
    connection.addRequestProperty("If-None-Match", mETag);
    connection.setRequestProperty("Connection", "Keep-Alive");
    connection.setRequestProperty("Content-Type", "application/json");
    connection.setRequestProperty("Charset", "UTF-8");
    connection.setRequestProperty("Content-Transfer-Encoding", "binary");
    connection.setRequestMethod("GET");
    connection.setDoInput(true);
    return connection;
}

private void subscribe(String url) {
    HttpURLConnection connection = null;
    while (!isCancelled()) {
        try {
            connection = getConnection(url);

            Log.i("", hashCode() + ": Requeting url: " + url + "; LastModify: " + mLastModify + "; ETAG: " + mETag);
            long begin_time = System.currentTimeMillis();
            int resCode = connection.getResponseCode();
            long end_time = System.currentTimeMillis();
            Log.i("", hashCode() + ": Response Code: " + resCode + "; Keep Alive: " + (end_time - begin_time) / 1000 + "(s)");
            if (resCode == 504) {
                throw new IOException("Gateway Time out");
            }

            if (resCode != 200) {
                throw new IOException("Response Status Code not 200");
            }

            InputStream ins = connection.getInputStream();
            parseInputstream(ins);
            ins.close();

            mLastModify = (String)connection.getHeaderField("last-modified");
            mETag = (String)connection.getHeaderField("etag");
        } catch (MalformedURLException e) {
            Log.w("", "MalformedURLException", e);
            break;
        } catch (IOException e) {
            Log.w("", "IOException", e);
            sleep(5);
            continue;
        } finally {
            if (connection != null) {
                connection.disconnect();
                connection = null;
            }
        }
    }
}

private void sleep(int seconds) {
    try {
        Thread.sleep(1000 * seconds);
    } catch (Exception e) {
    }
}

}

wandenberg commented 12 years ago

Hi,

sorry for the late response.

I'm not sure, but this problem may be occurring only on your application. I will need more information to try to help you. The module always send the chunk following the specification (chunk-size\r\nchunk-data\r\n) on the same write operation, there is no reason to be "intermittent".

In the code you send me the parseInputstream is missing, can you send me it? Send me your nginx.conf too, please.

And, can you try to reproduce the problem with nginx 1.2.0 version (I know that isn't the last stable, but was the last version I used on my tests and development, and it was fully functional)

I tried to use your server at notify.weikan.cn but received a 404 on /pub location.

Regards, Wandenberg

On Mon, Aug 20, 2012 at 11:57 PM, Terry.Mao notifications@github.comwrote:

i post msg use java code :

public static boolean post(String url, String msg) { HttpPost post = null; InputStreamEntity reqEntity = null; HttpEntity resEntity = null; BufferedReader http_reader = null; StringBuffer sResult = null;

try {
    post = new HttpPost(url);
    /*
     * reqEntity = new StringEntity(msg, "application/json", "UTF-8");
     * reqEntity.setContentEncoding(new BasicHeader(HTTP.CONTENT_TYPE,
     * "application/json")); reqEntity.setChunked(false);
     */

    byte[] reqData = msg.getBytes("UTF-8");
    reqEntity = new InputStreamEntity(
            new ByteArrayInputStream(reqData), reqData.length);
    reqEntity.setChunked(true);
    post.setEntity((HttpEntity) reqEntity);
    PushFeeder.logger.info("Content-Length : "
            + reqEntity.getContentLength());
    HttpResponse res = client.execute(post);

    if (HttpStatus.SC_OK == res.getStatusLine().getStatusCode()) {
        resEntity = res.getEntity();
        if (resEntity != null) {
            http_reader = new BufferedReader(new InputStreamReader(
                    resEntity.getContent()));

            String line = null;
            sResult = new StringBuffer();
            while ((line = http_reader.readLine()) != null) {
                if (sResult.length() > 0) {
                    sResult.append("\n");
                }
                sResult.append(line);
            }

            PushFeeder.logger.info(sResult.toString());
        }

        return true;
    } else {
        PushFeeder.logger.info("url : " + url + ", ret : "
                + res.getStatusLine().getStatusCode());
    }

} catch (Exception e) {
    PushFeeder.logger.error("", e);
} finally {
    if (http_reader != null) {
        try {
            http_reader.close();
        } catch (IOException e) {
            PushFeeder.logger.error("", e);
        }
    }
}

return false;

}

my android client use java code to get response:

public class TestSubcribe { private String mLastModify; private String mETag;

public void run() { String url = "http://****/PUBLIC"; subcribe(url); }

private HttpURLConnection getConnection(String url) throws MalformedURLException, IOException { HttpURLConnection connection = (HttpURLConnection)(new URL(url.toString()).openConnection()); connection.setConnectTimeout(30 * 60 * 1000); connection.setReadTimeout(30 * 60 * 1000); connection.addRequestProperty("If-Modified-Since", mLastModify); connection.addRequestProperty("If-None-Match", mETag); connection.setRequestProperty("Connection", "Keep-Alive"); connection.setRequestProperty("Content-Type", "application/json"); connection.setRequestProperty("Charset", "UTF-8"); connection.setRequestProperty("Content-Transfer-Encoding", "binary"); connection.setRequestMethod("GET"); connection.setDoInput(true); return connection; }

private void subscribe(String url) { HttpURLConnection connection = null; while (!isCancelled()) { try { connection = getConnection(url);

        Log.i("", hashCode() + ": Requeting url: " + url + "; LastModify: " + mLastModify + "; ETAG: " + mETag);
        long begin_time = System.currentTimeMillis();
        int resCode = connection.getResponseCode();
        long end_time = System.currentTimeMillis();
        Log.i("", hashCode() + ": Response Code: " + resCode + "; Keep Alive: " + (end_time - begin_time) / 1000 + "(s)");
        if (resCode == 504) {
            throw new IOException("Gateway Time out");
        }

        if (resCode != 200) {
            throw new IOException("Response Status Code not 200");
        }

        InputStream ins = connection.getInputStream();
        parseInputstream(ins);
        ins.close();

        mLastModify = (String)connection.getHeaderField("last-modified");
        mETag = (String)connection.getHeaderField("etag");
    } catch (MalformedURLException e) {
        Log.w("", "MalformedURLException", e);
        break;
    } catch (IOException e) {
        Log.w("", "IOException", e);
        sleep(5);
        continue;
    } finally {
        if (connection != null) {
            connection.disconnect();
            connection = null;
        }
    }
}

}

private void sleep(int seconds) { try { Thread.sleep(1000 * seconds); } catch (Exception e) { } }

}

— Reply to this email directly or view it on GitHubhttps://github.com/wandenberg/nginx-push-stream-module/issues/45#issuecomment-7890261.

MysqlProxy commented 12 years ago

I tried to use your server at notify.weikan.cn but received a 404 on /pub

oh, sorry. i only allow my Intranet access the /pub.

the nginx.conf :

user nobody nobody; worker_processes 8;

error_log /data/logs/nginx/error.log warn; pid /var/run/nginx.pid;

events { use epoll; worker_connections 81920; }

http { include /etc/nginx/mime.types; default_type application/octet-stream;

log_format  default '$remote_addr - $remote_user [$time_local] "$request" $host '
                  '$status $body_bytes_sent "$http_referer" '
                  '"$http_user_agent" "$http_x_forwarded_for"';

log_format main '$time_local $request_time $bytes_sent $host $request';
log_format getip '$host $server_addr $remote_addr $proxy_add_x_forwarded_for';

access_log  /data/logs/nginx/access.log main buffer=1M;

sendfile        on;
tcp_nopush     on;

keepalive_timeout  120;

gzip  on;
gzip_min_length  1024;
gzip_buffers     4 8k;
gzip_types       text/plain application/x-javascript text/css application/xml text/javascript application/json;

server {
    listen       80;
    charset utf-8;
    access_log  off;

    location / {
        root   /nfs/status;
        add_header id 22;
        access_log  off;
        index  index.html index.htm;
    }
    location = /status {
        stub_status on;
        access_log  off;
    }
}

include /etc/nginx/conf.d/*.conf;

}

the parseInputstream (now i can only handle the issue by match {} to parse json):

private void parseInputstream(InputStream ins) throws IOException { StringBuffer sb = new StringBuffer(); int MAX_BUFFER_SIZE = 1024 * 8; byte[] buffer = new byte[MAX_BUFFER_SIZE]; int left_bracket_count = 0; int index = 0; while (true) { int chr = ins.read(); if (chr < 0) break; // 到内容末尾了

            if (chr != 123 && left_bracket_count == 0) continue;    // 还没有找到字符 ‘{’,忽略该字符  

            buffer[index++] = (byte)chr;
            if (index >= MAX_BUFFER_SIZE) {
                sb.append(new String(buffer, 0, index));
                index = 0;
            }

            if (chr == 123) {   // '{'
                left_bracket_count ++;
            } else if (chr == 125) {    // '}'
                left_bracket_count --;
                if (left_bracket_count == 0) {
                    sb.append(new String(buffer, 0, index));
                    publish(sb.toString());
                    sb.setLength(0);    // clear StringBuffer
                }
            }
        }

        if (sb.length() > 0) {
            publish(sb.toString());
            sb.setLength(0);
        }
    }

allright,i'll try the nginx 1.2.0 version. thks a lot! you are so kind man!~~